1137 lines
25 KiB
Python
1137 lines
25 KiB
Python
# GEEN SQL IN MAIN.PY, DIT IS DE API/ERMISSIECHECK, HOU HET OVERZICHTELIJK!!
|
|
|
|
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, Field
|
|
from typing import Literal
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from fastapi import Query
|
|
|
|
from database import fetch_one, execute
|
|
from email_service import hash_token
|
|
from fastapi.responses import PlainTextResponse, HTMLResponse
|
|
|
|
from database import get_table_names
|
|
|
|
from email_service import (
|
|
update_email
|
|
)
|
|
|
|
from auth import (
|
|
create_user,
|
|
verify_user,
|
|
create_token,
|
|
get_current_user,
|
|
get_user,
|
|
change_password,
|
|
delete_me,
|
|
)
|
|
|
|
from permissions import (
|
|
has_role,
|
|
whoami,
|
|
get_role,
|
|
)
|
|
|
|
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,
|
|
update_member_role,
|
|
)
|
|
|
|
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,
|
|
confirm_event,
|
|
cancel_event,
|
|
)
|
|
|
|
from event_signups import (
|
|
respond_to_event,
|
|
count_confirmations,
|
|
get_event_responses,
|
|
)
|
|
|
|
class EmailUpdate(BaseModel):
|
|
email: EmailStr
|
|
|
|
class MemberRoleUpdate(BaseModel):
|
|
role: Literal["member", "admin"]
|
|
|
|
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
|
|
password: str
|
|
email: EmailStr
|
|
|
|
|
|
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
|
|
type: str | None = None
|
|
visibility: str | None = None
|
|
status: Literal["draft", "proposed", "confirmed", "cancelled"] | None = None
|
|
|
|
class RoomCreate(BaseModel):
|
|
room_name: str
|
|
invite_code: str = Field(
|
|
min_length=6,
|
|
max_length=32,
|
|
pattern=r"^[A-Za-z0-9]+$",
|
|
)
|
|
|
|
|
|
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(
|
|
"/",
|
|
summary="API status",
|
|
description="Returns the current status of the Calendar API.",
|
|
)
|
|
def home():
|
|
return {
|
|
"status": "online",
|
|
"service": "Calendar API",
|
|
}
|
|
|
|
|
|
@app.get(
|
|
"/database",
|
|
summary="Get database status",
|
|
description="Returns the names of all tables currently available in the 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",
|
|
summary="Create a user",
|
|
description="Creates a new user account and sends an email verification link. Usernames should be unique",
|
|
)
|
|
def api_create_user(user: UserCreate):
|
|
try:
|
|
user_id = create_user(
|
|
user.username,
|
|
user.password,
|
|
user.email,
|
|
)
|
|
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="username or email already exists",
|
|
)
|
|
|
|
return {
|
|
"status": "created",
|
|
"user_id": user_id,
|
|
"email_verification": "sent",
|
|
}
|
|
|
|
|
|
@app.post(
|
|
"/login",
|
|
summary="Log in",
|
|
description="Authenticates a user and returns a JWT access token.",
|
|
)
|
|
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",
|
|
summary="Get current user",
|
|
description="Returns the profile of the currently authenticated user.",
|
|
)
|
|
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)
|
|
|
|
class DeleteMeRequest(BaseModel):
|
|
password: str
|
|
|
|
|
|
@app.delete("/me")
|
|
def api_delete_me(
|
|
data: DeleteMeRequest,
|
|
user_id: int = Depends(get_current_user),
|
|
):
|
|
result = delete_me(user_id, data.password)
|
|
|
|
if result is False:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Incorrect password"
|
|
)
|
|
|
|
return {
|
|
"status": "deleted"
|
|
}
|
|
|
|
@app.get(
|
|
"/users/{user_id}",
|
|
summary="Get a user",
|
|
description="Returns the profile of a user by their 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.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",
|
|
description="Changes a user's role in a room. Only the room owner can edit member roles.",
|
|
)
|
|
def api_update_member_role(
|
|
room_id: int,
|
|
user_id: int,
|
|
data: MemberRoleUpdate,
|
|
current_user: int = Depends(get_current_user),
|
|
):
|
|
current_role = get_role(room_id, current_user)
|
|
target_role = get_role(room_id, user_id)
|
|
|
|
if current_role is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="not a room member",
|
|
)
|
|
|
|
if target_role is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="user is not a member of this room",
|
|
)
|
|
|
|
if current_role < 3:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="owner permission required",
|
|
)
|
|
|
|
if target_role >= current_role:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="cannot edit a member with equal or higher role",
|
|
)
|
|
|
|
update_member_role(
|
|
room_id,
|
|
user_id,
|
|
data.role,
|
|
)
|
|
|
|
return {
|
|
"status": "updated",
|
|
"user_id": user_id,
|
|
"role": data.role,
|
|
}
|
|
|
|
# rooms
|
|
|
|
@app.post(
|
|
"/create-room",
|
|
summary="Create a room",
|
|
description="Creates a new room and makes the authenticated user its owner.",
|
|
)
|
|
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",
|
|
summary="Search for a room",
|
|
description="Searches for a room using its invite code.",
|
|
)
|
|
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",
|
|
summary="Join a room",
|
|
description="Joins the authenticated user to a room using its invite code.",
|
|
)
|
|
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",
|
|
summary="Get user's rooms",
|
|
description="Returns all rooms the authenticated user is a member of.",
|
|
)
|
|
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}",
|
|
summary="Get a room",
|
|
description="Returns information about a room. The authenticated user must be a member.",
|
|
)
|
|
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",
|
|
summary="Get current user's room role",
|
|
description="Returns the authenticated user's role and permissions within a room.",
|
|
)
|
|
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",
|
|
summary="Get room members",
|
|
description="Returns all members of a room. The authenticated user must be a member.",
|
|
)
|
|
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",
|
|
summary="Leave a room",
|
|
description="Removes the authenticated user from a room.",
|
|
)
|
|
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}",
|
|
summary="Remove a room member",
|
|
description="""
|
|
Removes a member from a room.
|
|
|
|
Admins can remove members with a lower role.
|
|
Owners can remove members and admins.
|
|
A user cannot remove someone with an equal or higher role.
|
|
""",
|
|
)
|
|
def api_remove_member(
|
|
room_id: int,
|
|
user_id: int,
|
|
current_user: int = Depends(get_current_user),
|
|
):
|
|
current_role = get_role(room_id, current_user)
|
|
target_role = get_role(room_id, user_id)
|
|
|
|
if current_role is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="not a room member",
|
|
)
|
|
|
|
if target_role is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="user is not a member of this room",
|
|
)
|
|
|
|
if current_role < 2:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="admin permission required",
|
|
)
|
|
|
|
if target_role >= current_role:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="cannot remove a member with equal or higher role",
|
|
)
|
|
|
|
remove_room_member(room_id, user_id)
|
|
|
|
return {
|
|
"status": "removed",
|
|
"user_id": user_id,
|
|
}
|
|
|
|
|
|
# room settings
|
|
|
|
@app.patch(
|
|
"/rooms/{room_id}",
|
|
summary="Rename a room",
|
|
description="Changes the name of a room. Requires admin or owner permissions.",
|
|
)
|
|
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",
|
|
summary="Regenerate invite code",
|
|
description="Generates a new invite code for a room. Requires admin or owner permissions.",
|
|
)
|
|
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}",
|
|
summary="Delete a room",
|
|
description="Permanently deletes a room. Only the room owner can perform this action.",
|
|
)
|
|
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",
|
|
summary="Get room events",
|
|
description="Returns events belonging to a room, optionally filtered by date.",
|
|
)
|
|
def api_room_events(
|
|
room_id: int,
|
|
date_from: date | None = Query(
|
|
None,
|
|
alias="from",
|
|
description="Only return events starting on or after this date.",
|
|
),
|
|
date_to: date | None = Query(
|
|
None,
|
|
alias="to",
|
|
description="Only return events starting on or before this date.",
|
|
),
|
|
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,
|
|
date_from.isoformat() if date_from else None,
|
|
date_to.isoformat() if date_to else None,
|
|
)
|
|
|
|
return {
|
|
"events": [dict(event) for event in events],
|
|
}
|
|
|
|
@app.post(
|
|
"/events",
|
|
summary="Create an event",
|
|
description="Creates a new event in a room. The authenticated user must be a member of the room.",
|
|
)
|
|
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",
|
|
summary="Get user's events",
|
|
description="Returns events from rooms the authenticated user belongs to, optionally filtered by date.",
|
|
)
|
|
def get_events(
|
|
date_from: date | None = Query(
|
|
None,
|
|
alias="from",
|
|
description="Only return events starting on or after this date.",
|
|
),
|
|
date_to: date | None = Query(
|
|
None,
|
|
alias="to",
|
|
description="Only return events starting on or before this date.",
|
|
),
|
|
user_id: int = Depends(get_current_user),
|
|
):
|
|
events = get_user_events(
|
|
user_id,
|
|
date_from.isoformat() if date_from else None,
|
|
date_to.isoformat() if date_to else None,
|
|
)
|
|
|
|
return {
|
|
"events": [dict(event) for event in events],
|
|
}
|
|
|
|
@app.get(
|
|
"/events/{event_id}",
|
|
summary="Get an event",
|
|
description="Returns a specific event. The authenticated user must be a member of its room.",
|
|
)
|
|
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}/confirm",
|
|
summary="Confirm an event",
|
|
description="Manually confirms an event. Only the event creator or a room admin can perform this action.",
|
|
)
|
|
def api_confirm_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",
|
|
)
|
|
|
|
confirm_event(event_id)
|
|
|
|
return {
|
|
"status": "confirmed",
|
|
"event_id": event_id,
|
|
}
|
|
|
|
|
|
@app.post(
|
|
"/events/{event_id}/cancel",
|
|
summary="Cancel an event",
|
|
description="Cancels an event. Only the event creator or a room admin can perform this action.",
|
|
)
|
|
def api_cancel_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",
|
|
)
|
|
|
|
cancel_event(event_id)
|
|
|
|
return {
|
|
"status": "cancelled",
|
|
"event_id": event_id,
|
|
}
|
|
|
|
|
|
@app.post(
|
|
"/events/{event_id}/respond",
|
|
summary="Respond to an event",
|
|
description="Saves the authenticated user's response to an event and checks whether the event reached its confirmation threshold.",
|
|
)
|
|
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",
|
|
summary="Get event responses",
|
|
description="Returns all responses submitted by members for an event.",
|
|
)
|
|
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}",
|
|
summary="Update an event",
|
|
description="Updates an event. The event creator or a room admin can perform this action.",
|
|
)
|
|
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,
|
|
event.type,
|
|
event.visibility,
|
|
event.status,
|
|
)
|
|
|
|
return {
|
|
"status": "updated",
|
|
}
|
|
|
|
|
|
@app.delete(
|
|
"/events/{event_id}",
|
|
summary="Delete an event",
|
|
description="Deletes an event. The event creator or a room admin can perform this action.",
|
|
)
|
|
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",
|
|
}
|
|
|
|
# email endpoints
|
|
|
|
@app.put(
|
|
"/me/email",
|
|
summary="Update email address",
|
|
description="Changes the authenticated user's email address.",
|
|
)
|
|
def api_update_email(
|
|
data: EmailUpdate,
|
|
user_id: int = Depends(get_current_user),
|
|
):
|
|
return update_email(
|
|
user_id,
|
|
data.email,
|
|
)
|
|
|
|
@app.get(
|
|
"/verify-email",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def api_verify_email(token: str = Query(...)):
|
|
result = verify_email(token)
|
|
|
|
if result["status"] == "failed":
|
|
if result["reason"] == "token_expired":
|
|
return """
|
|
<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<title>Verification expired</title>
|
|
</head>
|
|
<body>
|
|
<h1>Verification expired</h1>
|
|
<p>This verification link has expired.</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return """
|
|
<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<title>Verification failed</title>
|
|
</head>
|
|
<body>
|
|
<h1>Verification failed</h1>
|
|
<p>This verification link is invalid.</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return """
|
|
<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>Email verified</title>
|
|
</head>
|
|
<body>
|
|
<div class="box">
|
|
<h1>Email verified</h1>
|
|
<p>Your email address has been successfully verified.</p>
|
|
<p>You may now close this tab.</p>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
# easter egg
|
|
|
|
@app.get(
|
|
"/ascii-art",
|
|
summary="Display ASCII art",
|
|
description="Returns the Calendar API ASCII art as plain text.",
|
|
response_class=PlainTextResponse,
|
|
)
|
|
def ascii_art():
|
|
return """ __ .__ __
|
|
___.__. ____ __ __ _____ _______ ____ _____ _______/ |_|__| ____ | | _____.__. ______ ____ ____
|
|
< | |/ _ \| | \ \__ \\\\_ __ \_/ __ \ \__ \ / ___/\ __\ |/ \| |/ < | | \____ \ / _ \ / _ \
|
|
\___ ( <_> ) | / / __ \| | \/\ ___/ / __ \_ \___ \ | | | | | \ < \___ | | |_> > <_> | <_> )
|
|
/ ____|\____/|____/ (____ /__| \___ > (____ / /____ > |__| |__|___| /__|_ \/ ____| | __/ \____/ \____/
|
|
\/ \/ \/ \/ \/ \/ \/\/ |__|
|
|
""" |