This commit is contained in:
ben
2026-08-18 14:27:52 +02:00
parent edada21b4d
commit 54581e2e42
3 changed files with 114 additions and 321 deletions
+47 -1
View File
@@ -104,4 +104,50 @@ def get_user(user_id):
WHERE id = ?
""",
(user_id,)
)
)
def change_password(user_id: int, old_password: str, new_password: str):
user = fetch_one(
"""
SELECT password_hash
FROM users
WHERE id = ?
""",
(user_id,),
)
if user is None:
return {
"status": "failed",
"reason": "user_not_found",
}
if not bcrypt.checkpw(
old_password.encode(),
user["password_hash"].encode(),
):
return {
"status": "failed",
"reason": "incorrect_password",
}
password_hash = bcrypt.hashpw(
new_password.encode(),
bcrypt.gensalt(),
).decode()
execute(
"""
UPDATE users
SET password_hash = ?
WHERE id = ?
""",
(
password_hash,
user_id,
),
)
return {
"status": "updated",
}
+41 -10
View File
@@ -4,7 +4,7 @@ from datetime import date
from fastapi import FastAPI, Depends, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel, EmailStr
from pydantic import BaseModel, EmailStr, Field
from typing import Literal
import sqlite3
from datetime import datetime, timezone
@@ -20,17 +20,13 @@ from email_service import (
update_email
)
from email_service import (
update_email,
verify_email,
)
from auth import (
create_user,
verify_user,
create_token,
get_current_user,
get_user,
change_password,
)
from permissions import (
@@ -84,6 +80,9 @@ class Login(BaseModel):
username: str
password: str
class PasswordUpdate(BaseModel):
old_password: str
new_password: str = Field(min_length=8)
class UserCreate(BaseModel):
username: str
@@ -113,10 +112,13 @@ class EventUpdate(BaseModel):
visibility: str | None = None
status: Literal["draft", "proposed", "confirmed", "cancelled"] | None = None
class RoomCreate(BaseModel):
room_name: str
invite_code: str
invite_code: str = Field(
min_length=6,
max_length=32,
pattern=r"^[A-Za-z0-9]+$",
)
class RoomSearch(BaseModel):
@@ -266,6 +268,35 @@ def api_get_user(
return dict(user)
@app.patch(
"/me/password",
summary="Change password",
description="Changes the authenticated user's password after verifying the current password.",
)
def api_change_password(
data: PasswordUpdate,
user_id: int = Depends(get_current_user),
):
result = change_password(
user_id,
data.old_password,
data.new_password,
)
if result["status"] == "failed":
if result["reason"] == "incorrect_password":
raise HTTPException(
status_code=401,
detail="incorrect password",
)
raise HTTPException(
status_code=404,
detail="user not found",
)
return result
@app.patch(
"/rooms/{room_id}/members/{user_id}",
summary="Edit a user's role",
@@ -528,8 +559,8 @@ def api_remove_member(
user_id: int,
current_user: int = Depends(get_current_user),
):
current_role = get_member_role(room_id, current_user)
target_role = get_member_role(room_id, user_id)
current_role = get_role(room_id, current_user)
target_role = get_role(room_id, user_id)
if current_role is None:
raise HTTPException(
+26 -310
View File
@@ -1,10 +1,29 @@
from fastapi import FastAPI
from pydantic import BaseModel
import re
import secrets
import string
from database import execute, fetch_one, fetch_all
DEFAULT_ROLE = "member"
def create_room(user_id, room_name, invite_code):
def validate_invite_code(invite_code: str):
if not 6 <= len(invite_code) <= 32:
return False
if not re.fullmatch(r"[A-Za-z0-9]+", invite_code):
return False
return True
def create_room(user_id: int, room_name: str, invite_code: str):
if not validate_invite_code(invite_code):
return {
"status": "failed",
"reason": "invalid_invite_code",
}
room_id = execute(
"""
@@ -14,7 +33,7 @@ def create_room(user_id, room_name, invite_code):
(
room_name,
user_id,
invite_code
invite_code,
)
)
@@ -26,314 +45,11 @@ def create_room(user_id, room_name, invite_code):
(
room_id,
user_id,
"owner"
"owner",
)
)
return {
"room_id": room_id,
"member_id": member_id
}
def search_room(
invite_code
):
query = """
SELECT name
FROM rooms
WHERE invite_code = ?
"""
room = fetch_one(
query,
(
invite_code,
)
)
if room is None:
return None
return room["name"]
def accept_room(user_id, invite_code):
query = """
SELECT id
FROM rooms
WHERE invite_code = ?
"""
room = fetch_one(
query,
(invite_code,)
)
if room is None:
return None
room_id = room["id"]
existing = fetch_one(
"""
SELECT *
FROM room_members
WHERE room_id = ?
AND user_id = ?
""",
(
room_id,
user_id
)
)
if existing:
return None
query = """
INSERT INTO room_members
(room_id, user_id, role)
VALUES (?, ?, ?)
"""
member_id = execute(
query,
(
room_id,
user_id,
"member"
)
)
return {
"room_id": room_id,
"member_id": member_id
}
def get_rooms(user_id):
query = """
SELECT
rooms.id,
rooms.name,
rooms.owner_id,
rooms.invite_code,
room_members.role
FROM rooms
JOIN room_members
ON rooms.id = room_members.room_id
WHERE room_members.user_id = ?
"""
rooms = fetch_all(
query,
(
user_id,
)
)
return rooms
def delete_room(room_id):
room = execute(
"""
SELECT id
FROM rooms
WHERE id = ?
""",
(room_id,)
)
if not room:
return {"error": "room not found"}
execute(
"""
DELETE FROM rooms
WHERE id = ?
""",
(room_id,)
)
return {"status": "deleted"}
def get_room(room_id, user_id):
room = fetch_one(
"""
SELECT
rooms.id,
rooms.name,
rooms.owner_id,
room_members.role
FROM rooms
JOIN room_members
ON rooms.id = room_members.room_id
WHERE rooms.id = ?
AND room_members.user_id = ?
""",
(
room_id,
user_id
)
)
if room is None:
return None
members = fetch_all(
"""
SELECT
users.id,
users.username,
room_members.role
FROM room_members
JOIN users
ON users.id = room_members.user_id
WHERE room_members.room_id = ?
""",
(room_id,)
)
events = fetch_all(
"""
SELECT *
FROM events
WHERE room_id = ?
ORDER BY start_time
""",
(room_id,)
)
return {
"room": dict(room),
"members": [dict(member) for member in members],
"events": [dict(event) for event in events]
}
def get_room_members(room_id):
return fetch_all(
"""
SELECT
users.id,
users.username
FROM room_members
JOIN users
ON users.id = room_members.user_id
WHERE room_members.room_id = ?
""",
(room_id,)
)
def leave_room(room_id, user_id):
from database import execute
execute(
"""
DELETE FROM room_members
WHERE room_id = ?
AND user_id = ?
""",
(
room_id,
user_id
)
)
def change_room_name(room_id, new_name, user_id):
room = fetch_one(
"""
SELECT id
FROM rooms
WHERE id = ?
""",
(room_id,)
)
if not room:
return {
"status": "error",
"reason": "room_not_found"
}
execute(
"""
UPDATE rooms
SET name = ?
WHERE id = ?
""",
(
new_name,
room_id
)
)
return {
"status": "updated",
"new_name": new_name,
"room_id": room_id,
"updated_by": user_id
}
import secrets
import string
def regenerate_invite_code(room_id):
new_code = ''.join(
secrets.choice(string.ascii_uppercase + string.digits)
for _ in range(8)
)
execute(
"""
UPDATE rooms
SET invite_code = ?
WHERE id = ?
""",
(new_code, room_id)
)
return {
"status": "updated",
"room_id": room_id,
"invite_code": new_code
}
def remove_room_member(room_id, user_id):
member = fetch_one(
"""
SELECT 1
FROM room_members
WHERE room_id = ? AND user_id = ?
""",
(room_id, user_id)
)
if member is None:
return False
execute(
"""
DELETE FROM room_members
WHERE room_id = ? AND user_id = ?
""",
(room_id, user_id)
)
return True
def update_member_role(room_id, user_id, role):
execute(
"""
UPDATE room_members
SET role = ?
WHERE room_id = ?
AND user_id = ?
""",
(role, room_id, user_id),
)
"member_id": member_id,
}