# GEEN SQL IN MAIN.PY, DIT IS DE API WRAPPER, HOU HET OVERZICHTELIJK!! from fastapi import FastAPI, Depends, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import PlainTextResponse from pydantic import BaseModel from typing import Literal import sqlite3 from database import get_table_names from auth import ( create_user, verify_user, create_token, get_current_user, get_user, ) from permissions import ( has_role, whoami, ) from rooms import ( create_room, search_room, accept_room, get_rooms, get_room, get_room_members, leave_room, delete_room as delete_room_db, change_room_name, regenerate_invite_code, remove_room_member, ) from events import ( create_event, get_user_events, check_event_threshold, delete_event, get_room_events, update_event, get_event, get_event_room_id, get_event_creator, ) from event_signups import ( respond_to_event, count_confirmations, get_event_responses, ) class Login(BaseModel): username: str password: str class UserCreate(BaseModel): username: str password: str class EventCreate(BaseModel): room_id: int type: str title: str description: str start_time: str end_time: str visibility: str status: str 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 class RoomSearch(BaseModel): invite_code: str class RoomJoin(BaseModel): invite_code: str class RoomUpdate(BaseModel): new_name: str app = FastAPI( title="Calendar API", description="Room based agenda system", version="0.2.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # generaal @app.get("/") def home(): return { "status": "online", "service": "Calendar API", } @app.get("/database") def database_status( user_id: int = Depends(get_current_user), ): tables = get_table_names() return { "tables": [table["name"] for table in tables], } # users @app.post("/create-user") def api_create_user(user: UserCreate): try: user_id = create_user( user.username, user.password, ) except sqlite3.IntegrityError: raise HTTPException( status_code=409, detail="username already exists", ) return { "status": "created", "user_id": user_id, } @app.post("/login") def api_login(login: Login): user_id = verify_user( login.username, login.password, ) if user_id is None: return { "status": "failed", } token = create_token(user_id) return { "status": "success", "access_token": token, "token_type": "bearer", } @app.get("/me") def get_me( 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", ) return dict(user) @app.get("/users/{user_id}") def api_get_user( user_id: int, 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", ) return dict(user) @app.post("/create-room") def api_create_room( room: RoomCreate, user_id: int = Depends(get_current_user), ): try: result = create_room( user_id, room.room_name, room.invite_code, ) except sqlite3.IntegrityError: raise HTTPException( status_code=409, detail="invite_code or roomname already exists", ) return { "status": "created", **result, } @app.post("/search-room") def api_search_room( room: RoomSearch, user_id: int = Depends(get_current_user), ): room_name = search_room( room.invite_code, ) if room_name is None: return { "status": "failed", "reason": "room_not_found", } return { "status": "found", "room_name": room_name, } @app.post("/join-room") def api_join_room( room: RoomJoin, user_id: int = Depends(get_current_user), ): result = accept_room( user_id, room.invite_code, ) if result is None: return { "status": "failed", "reason": "room_not_found_or_already_member", } return { "status": "joined", **result, } @app.get("/rooms") def api_get_rooms( user_id: int = Depends(get_current_user), ): rooms = get_rooms(user_id) return { "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), ): 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, ) if room is None: raise HTTPException( status_code=404, detail="room not found", ) return dict(room) @app.get("/rooms/{room_id}/whoami") def api_room_whoami( room_id: int, user_id: int = Depends(get_current_user), ): permissions = whoami(room_id, user_id) if not permissions["member"]: raise HTTPException( status_code=403, detail="not a room member", ) return { "room_id": room_id, "user_id": user_id, **permissions, } @app.get("/rooms/{room_id}/members") def api_room_members( 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="room member permission required", ) members = get_room_members(room_id) return { "members": [dict(member) for member in members], } @app.delete("/rooms/{room_id}/members/me") def api_leave_room( 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", ) leave_room( room_id, user_id, ) return { "status": "left", } @app.delete("/rooms/{room_id}/members/{user_id}") def remove_user_as_admin( room_id: int, user_id: int, current_user: int = Depends(get_current_user), ): if not has_role(room_id, current_user, "admin"): raise HTTPException( status_code=403, detail="admin permission required", ) if not remove_room_member(room_id, user_id): raise HTTPException( status_code=404, detail="user is not a member of this room", ) return { "status": "removed", "user_id": user_id, } # some settings @app.patch("/rooms/{room_id}") def update_room( room_id: int, data: RoomUpdate, 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", ) return change_room_name( room_id, data.new_name, user_id, ) @app.post("/rooms/{room_id}/invite-code") def update_invite_code( room_id: int, 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", ) return regenerate_invite_code(room_id) @app.delete("/rooms/{room_id}") def delete_room( room_id: int, 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", ) return delete_room_db(room_id) # 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="room member permission required", ) events = get_room_events(room_id) return { "events": [dict(event) for event in events], } @app.post("/events") def api_create_event( event: EventCreate, 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", ) event_id = create_event( room_id=event.room_id, creator_id=user_id, type=event.type, title=event.title, description=event.description, start_time=event.start_time, end_time=event.end_time, visibility=event.visibility, status=event.status, min_people=event.min_people, ) return { "status": "created", "event_id": event_id, } @app.get("/events") def get_events( user_id: int = Depends(get_current_user), ): events = get_user_events(user_id) return { "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), ): event = get_event(event_id) if event is None: raise HTTPException( status_code=404, detail="event not found", ) if not has_role( event["room_id"], user_id, "member", ): raise HTTPException( status_code=403, 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), ): room_id = get_event_room_id(event_id) if room_id is None: raise HTTPException( status_code=404, detail="event not found", ) if not has_role( room_id, user_id, "member", ): raise HTTPException( status_code=403, detail="room member permission required", ) respond_to_event( event_id, user_id, status, ) check_event_threshold(event_id) return { "status": "response_saved", "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), ): room_id = get_event_room_id(event_id) if room_id is None: raise HTTPException( status_code=404, detail="event not found", ) if not has_role( room_id, user_id, "member", ): raise HTTPException( status_code=403, detail="room member permission required", ) responses = get_event_responses(event_id) return { "responses": [ dict(response) for response in responses ], } @app.patch("/events/{event_id}") def api_update_event( event_id: int, event: EventUpdate, user_id: int = Depends(get_current_user), ): room_id = get_event_room_id(event_id) creator_id = get_event_creator(event_id) if room_id is None or creator_id is None: raise HTTPException( status_code=404, detail="event not found", ) 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 creator or admin permission required", ) update_event( event_id, event.title, event.description, event.start_time, event.end_time, event.min_people, ) return { "status": "updated", } @app.delete("/events/{event_id}") def api_delete_event( event_id: int, user_id: int = Depends(get_current_user), ): room_id = get_event_room_id(event_id) creator_id = get_event_creator(event_id) if room_id is None or creator_id is None: raise HTTPException( status_code=404, detail="event not found", ) 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 creator or admin permission required", ) delete_event(event_id) return { "status": "deleted", } #easter egg @app.get( "/ascii-art", response_class=PlainTextResponse, ) def ascii_art(): return """ ██████╗ █████╗ ██╗ ███████╗███╗ ██╗██████╗ █████╗ ██████╗ ██╔════╝██╔══██╗██║ ██╔════╝████╗ ██║██╔══██╗██╔══██╗██╔══██╗ ██║ ███████║██║ █████╗ ██╔██╗ ██║██║ ██║███████║██████╔╝ ██║ ██╔══██║██║ ██╔══╝ ██║╚██╗██║██║ ██║██╔══██║██╔══██╗ ╚██████╗██║ ██║███████╗███████╗██║ ╚████║██████╔╝██║ ██║██║ ██║ ╚═════╝╚═╝ ╚═╝╚══════╝╚══════╝╚═╝ ╚═══╝╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝ status: online "You are a stinky Poo" - Ben de Roo, programmer """