From d47a46c81e945333157e7945547c5edf74fdd1d5 Mon Sep 17 00:00:00 2001 From: ben Date: Mon, 17 Aug 2026 16:08:13 +0200 Subject: [PATCH] restructured --- backend/main.py | 355 ++++++++++++++++++++++------------------- backend/permissions.py | 45 ++++-- 2 files changed, 228 insertions(+), 172 deletions(-) diff --git a/backend/main.py b/backend/main.py index a5ec840..0ae9ba1 100644 --- a/backend/main.py +++ b/backend/main.py @@ -5,16 +5,20 @@ from pydantic import BaseModel from typing import Literal import sqlite3 -from database import fetch_all, fetch_one +from database import get_table_names from auth import ( create_user, verify_user, create_token, - get_current_user + get_current_user, + get_user, ) -from permissions import has_role +from permissions import ( + has_role, + whoami, +) from rooms import ( create_room, @@ -26,31 +30,7 @@ from rooms import ( leave_room, delete_room as delete_room_db, change_room_name, - regenerate_invite_code -) - -from events import ( - create_event, - get_user_events, - check_event_threshold, - delete_event, - get_room_events, - update_event -) - -from event_signups import ( - respond_to_event, - count_confirmations -) - -from database import get_table_names - -from auth import ( - create_user, - verify_user, - create_token, - get_current_user, - get_user + regenerate_invite_code, ) from events import ( @@ -62,16 +42,15 @@ from events import ( update_event, get_event, get_event_room_id, - get_event_creator + get_event_creator, ) from event_signups import ( respond_to_event, count_confirmations, - get_event_responses + get_event_responses, ) - class Login(BaseModel): username: str password: str @@ -94,6 +73,14 @@ class EventCreate(BaseModel): min_people: int +class EventUpdate(BaseModel): + title: str | None = None + description: str | None = None + start_time: str | None = None + end_time: str | None = None + min_people: int | None = None + + class RoomCreate(BaseModel): room_name: str invite_code: str @@ -111,18 +98,10 @@ class RoomUpdate(BaseModel): new_name: str -class EventUpdate(BaseModel): - title: str | None = None - description: str | None = None - start_time: str | None = None - end_time: str | None = None - min_people: int | None = None - - app = FastAPI( title="Calendar API", description="Room based agenda system", - version="1.0" + version="1.0", ) @@ -134,58 +113,59 @@ app.add_middleware( allow_headers=["*"], ) -# general +# generaal @app.get("/") def home(): return { "status": "online", - "service": "Calendar API" + "service": "Calendar API", } @app.get("/database") -def database_status(): +def database_status( + user_id: int = Depends(get_current_user), +): tables = get_table_names() return { - "tables": [table["name"] for table in tables] + "tables": [table["name"] for table in tables], } -# user endpoints + +# users @app.post("/create-user") def api_create_user(user: UserCreate): - try: user_id = create_user( user.username, - user.password + user.password, ) except sqlite3.IntegrityError: raise HTTPException( status_code=409, - detail="username already exists" + detail="username already exists", ) return { "status": "created", - "user_id": user_id + "user_id": user_id, } @app.post("/login") def api_login(login: Login): - user_id = verify_user( login.username, - login.password + login.password, ) if user_id is None: return { - "status": "failed" + "status": "failed", } token = create_token(user_id) @@ -193,263 +173,282 @@ def api_login(login: Login): return { "status": "success", "access_token": token, - "token_type": "bearer" + "token_type": "bearer", } @app.get("/me") def get_me( - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): user = get_user(user_id) if user is None: raise HTTPException( status_code=404, - detail="user not found" + detail="user not found", ) return dict(user) + @app.get("/users/{user_id}") def api_get_user( user_id: int, - current_user: int = Depends(get_current_user) + current_user: int = Depends(get_current_user), ): user = get_user(user_id) if user is None: raise HTTPException( status_code=404, - detail="user not found" + detail="user not found", ) return dict(user) -# room endpoints @app.post("/create-room") def api_create_room( room: RoomCreate, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - try: result = create_room( user_id, room.room_name, - room.invite_code + room.invite_code, ) except sqlite3.IntegrityError: raise HTTPException( status_code=409, - detail="invite_code or roomname already exists" + detail="invite_code or roomname already exists", ) return { "status": "created", - **result + **result, } @app.post("/search-room") -def api_search_room(room: RoomSearch): - +def api_search_room( + room: RoomSearch, + user_id: int = Depends(get_current_user), +): room_name = search_room( - room.invite_code + room.invite_code, ) if room_name is None: return { "status": "failed", - "reason": "room_not_found" + "reason": "room_not_found", } return { "status": "found", - "room_name": room_name + "room_name": room_name, } @app.post("/join-room") def api_join_room( room: RoomJoin, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - result = accept_room( user_id, - room.invite_code + room.invite_code, ) if result is None: return { "status": "failed", - "reason": "room_not_found_or_already_member" + "reason": "room_not_found_or_already_member", } return { "status": "joined", - **result + **result, } @app.get("/rooms") def api_get_rooms( - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - rooms = get_rooms(user_id) return { - "rooms": [dict(room) for room in rooms] + "rooms": [dict(room) for room in rooms], } @app.get("/rooms/{room_id}") def api_get_room( room_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): + if not has_role(room_id, user_id, "member"): + raise HTTPException( + status_code=403, + detail="room member permission required", + ) room = get_room( room_id, - user_id + user_id, ) if room is None: raise HTTPException( status_code=404, - detail="room not found or no access" + detail="room not found", ) - return room + return dict(room) + + +@app.get("/rooms/{room_id}/whoami") +def api_room_whoami( + room_id: int, + user_id: int = Depends(get_current_user), +): + role = whoami( + room_id, + user_id, + ) + + if role is None: + raise HTTPException( + status_code=403, + detail="not a room member", + ) + + return { + "room_id": room_id, + "user_id": user_id, + "role": role, + } @app.get("/rooms/{room_id}/members") def api_room_members( room_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(room_id, user_id, "member"): raise HTTPException( status_code=403, - detail="room member permission required" + detail="room member permission required", ) members = get_room_members(room_id) return { - "members": [dict(member) for member in members] + "members": [dict(member) for member in members], } -@app.get("/rooms/{room_id}/events") -def api_room_events( +@app.delete("/rooms/{room_id}/members/me") +def api_leave_room( room_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(room_id, user_id, "member"): raise HTTPException( status_code=403, - detail="room member permission required" + detail="not a room member", ) - events = get_room_events(room_id) + leave_room( + room_id, + user_id, + ) return { - "events": [ - dict(event) - for event in events - ] + "status": "left", } -# settings +# some settings @app.patch("/rooms/{room_id}") def update_room( room_id: int, data: RoomUpdate, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(room_id, user_id, "admin"): raise HTTPException( status_code=403, - detail="admin permission required" + detail="admin permission required", ) return change_room_name( room_id, data.new_name, - user_id + user_id, ) @app.post("/rooms/{room_id}/invite-code") def update_invite_code( room_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(room_id, user_id, "admin"): raise HTTPException( status_code=403, - detail="admin permission required" + detail="admin permission required", ) return regenerate_invite_code(room_id) + @app.delete("/rooms/{room_id}") def delete_room( room_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(room_id, user_id, "owner"): raise HTTPException( status_code=403, - detail="owner permission required" + detail="owner permission required", ) return delete_room_db(room_id) -@app.delete("/rooms/{room_id}/members/me") -def api_leave_room( - room_id: int, - user_id: int = Depends(get_current_user) -): +# events + +@app.get("/rooms/{room_id}/events") +def api_room_events( + room_id: int, + user_id: int = Depends(get_current_user), +): if not has_role(room_id, user_id, "member"): raise HTTPException( status_code=403, - detail="not a room member" + detail="room member permission required", ) - leave_room( - room_id, - user_id - ) + events = get_room_events(room_id) return { - "status": "left" + "events": [dict(event) for event in events], } -# events - @app.post("/events") def api_create_event( event: EventCreate, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - if not has_role(event.room_id, user_id, "member"): raise HTTPException( status_code=403, - detail="room member permission required" + detail="room member permission required", ) event_id = create_event( @@ -462,126 +461,152 @@ def api_create_event( end_time=event.end_time, visibility=event.visibility, status=event.status, - min_people=event.min_people + min_people=event.min_people, ) return { "status": "created", - "event_id": event_id + "event_id": event_id, } @app.get("/events") def get_events( - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): - events = get_user_events(user_id) return { - "events": [dict(event) for event in events] + "events": [dict(event) for event in events], } @app.get("/events/{event_id}") def api_get_event( event_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): event = get_event(event_id) if event is None: raise HTTPException( status_code=404, - detail="event not found" + detail="event not found", ) - if not has_role(event["room_id"], user_id, "member"): + if not has_role( + event["room_id"], + user_id, + "member", + ): raise HTTPException( status_code=403, - detail="room member permission required" + detail="room member permission required", ) return dict(event) + @app.post("/events/{event_id}/respond") def respond_event( event_id: int, - status: Literal["going", "maybe", "declined"], - user_id: int = Depends(get_current_user) + status: Literal[ + "going", + "maybe", + "declined", + ], + user_id: int = Depends(get_current_user), ): room_id = get_event_room_id(event_id) if room_id is None: raise HTTPException( status_code=404, - detail="event not found" + detail="event not found", ) - if not has_role(room_id, user_id, "member"): + if not has_role( + room_id, + user_id, + "member", + ): raise HTTPException( status_code=403, - detail="room member permission required" + detail="room member permission required", ) respond_to_event( event_id, user_id, - status + status, ) check_event_threshold(event_id) return { "status": "response_saved", - "going": count_confirmations(event_id) + "going": count_confirmations(event_id), } + @app.get("/events/{event_id}/responses") def api_event_responses( event_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): room_id = get_event_room_id(event_id) if room_id is None: raise HTTPException( status_code=404, - detail="event not found" + detail="event not found", ) - if not has_role(room_id, user_id, "member"): + if not has_role( + room_id, + user_id, + "member", + ): raise HTTPException( status_code=403, - detail="room member permission required" + detail="room member permission required", ) responses = get_event_responses(event_id) return { - "responses": [dict(response) for response in responses] + "responses": [ + dict(response) + for response in responses + ], } -# more events - @app.patch("/events/{event_id}") def api_update_event( event_id: int, event: EventUpdate, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): + room_id = get_event_room_id(event_id) creator_id = get_event_creator(event_id) - if creator_id is None: + if room_id is None or creator_id is None: raise HTTPException( status_code=404, - detail="event not found" + detail="event not found", ) - if creator_id != user_id: + is_creator = creator_id == user_id + is_admin = has_role( + room_id, + user_id, + "admin", + ) + + if not is_creator and not is_admin: raise HTTPException( status_code=403, - detail="event owner permission required" + detail="event creator or admin permission required", ) update_event( @@ -590,44 +615,54 @@ def api_update_event( event.description, event.start_time, event.end_time, - event.min_people + event.min_people, ) return { - "status": "updated" + "status": "updated", } @app.delete("/events/{event_id}") def api_delete_event( event_id: int, - user_id: int = Depends(get_current_user) + user_id: int = Depends(get_current_user), ): + room_id = get_event_room_id(event_id) creator_id = get_event_creator(event_id) - if creator_id is None: + if room_id is None or creator_id is None: raise HTTPException( status_code=404, - detail="event not found" + detail="event not found", ) - if creator_id != user_id: + is_creator = creator_id == user_id + is_admin = has_role( + room_id, + user_id, + "admin", + ) + + if not is_creator and not is_admin: raise HTTPException( status_code=403, - detail="event owner permission required" + detail="event creator or admin permission required", ) delete_event(event_id) return { - "status": "deleted" + "status": "deleted", } -# easter egg +#easter egg -@app.get("/ascii-art", response_class=PlainTextResponse) +@app.get( + "/ascii-art", + response_class=PlainTextResponse, +) def ascii_art(): - return """ ██████╗ █████╗ ██╗ ███████╗███╗ ██╗██████╗ █████╗ ██████╗ ██╔════╝██╔══██╗██║ ██╔════╝████╗ ██║██╔══██╗██╔══██╗██╔══██╗ diff --git a/backend/permissions.py b/backend/permissions.py index f7c7035..6f61002 100644 --- a/backend/permissions.py +++ b/backend/permissions.py @@ -1,7 +1,14 @@ from database import fetch_one -def get_room_role(room_id, user_id): +ROLE_LEVELS = { + "member": 1, + "admin": 2, + "owner": 3, +} + + +def get_role(room_id: int, user_id: int): member = fetch_one( """ SELECT role @@ -9,10 +16,7 @@ def get_room_role(room_id, user_id): WHERE room_id = ? AND user_id = ? """, - ( - room_id, - user_id - ) + (room_id, user_id) ) if member is None: @@ -21,14 +25,31 @@ def get_room_role(room_id, user_id): return member["role"] -def has_role(room_id, user_id, allowed_roles): - - role = get_room_role( - room_id, - user_id - ) +def has_role( + room_id: int, + user_id: int, + required_role: str +) -> bool: + role = get_role(room_id, user_id) if role is None: return False - return role in allowed_roles \ No newline at end of file + return ROLE_LEVELS.get(role, 0) >= ROLE_LEVELS[required_role] + + +def whoami(room_id: int, user_id: int): + role = get_role(room_id, user_id) + + if role is None: + return { + "member": False, + "role": None, + } + + return { + "member": True, + "role": role, + "is_admin": ROLE_LEVELS[role] >= ROLE_LEVELS["admin"], + "is_owner": role == "owner", + } \ No newline at end of file