restructured

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
 
-@app.get("/ascii-art", response_class=PlainTextResponse)
+@app.get(
+    "/ascii-art",
+    response_class=PlainTextResponse,
+)
 def ascii_art():
-
     return """
      ██████╗ █████╗ ██╗     ███████╗███╗   ██╗██████╗  █████╗ ██████╗
     ██╔════╝██╔══██╗██║     ██╔════╝████╗  ██║██╔══██╗██╔══██╗██╔══██╗

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
+    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",
+    }