backend/backend/app.py

from datetime import datetime
from typing import Literal
from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel
import pymongo

from .db import MongoModel, PyObjectId, client
from .settings import settings

db = client["party"]
meta = client["party-meta"]

description = """
Party party \U0001F973
"""

tags_metadata = [
    {
        "name": "guests",
        "description": "Operations with guests (i.e., users). Intended to be called from the respective party frontend.",
    },
    {
        "name": "admin",
        "description": "Operations for administrative purposes. Require the **admin token**. Intended to be called from the admin UI",
    },
]

app = FastAPI(
    title="PartyPage Manager",
    description=description,
    openapi_tags=tags_metadata,
)

if settings.cors_origins or settings.cors_regex:
    from fastapi.middleware.cors import CORSMiddleware

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_origin_regex=settings.cors_regex,
        allow_methods=["*"],
        allow_headers=["*"],
    )

Coming = Literal["yes", "no", "maybe"]
GrammaticalGender = Literal["m", "f", "d"]


class HTTPError(BaseModel):
    detail: str


# error_responses = {status.HTTP_401_UNAUTHORIZED: {"model": HTTPError}}
def error_responses(*args):
    return {arg: {"model": HTTPError} for arg in args}


class Guest(BaseModel):
    token: str
    name: str
    coming: Coming | None
    grammatical_gender: GrammaticalGender
    extra: dict[str, str]


class DBGuest(Guest, MongoModel):
    pass


async def find_guest(party: str, token: str) -> DBGuest:
    guest = await db[party].find_one({"token": token})
    if not guest:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
    return DBGuest.parse_obj(guest)


class Party(MongoModel):
    name: str
    created: datetime
    allowed_extra: list[str]


async def find_party(name: str) -> Party:
    party = await meta["parties"].find_one({"name": name})
    if not party:
        raise HTTPException(status.HTTP_404_NOT_FOUND)
    return Party.parse_obj(party)


def validate_extra(extra: dict[str, str], party: Party):
    return all(k in party.allowed_extra and len(v) <= 64 for (k, v) in extra.items())


# Guest methods


@app.get(
    "/{party}/{token}/me",
    response_model=Guest,
    responses=error_responses(401),
    tags=["guests"],
)
async def get_self(guest: DBGuest = Depends(find_guest)):
    return guest


class GuestUpdate(BaseModel):
    coming: Coming | None
    extra: dict[str, str] | None


@app.patch(
    "/{party}/{token}/me",
    response_model=Guest,
    responses=error_responses(401),
    tags=["guests"],
)
async def update_self(
    party: str, update: GuestUpdate, guest: DBGuest = Depends(find_guest)
):
    try:
        party_obj = await find_party(party)
    except HTTPException:
        # should not happen since find_guest in Depends already
        # implies that the party/token combo is correct
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    guest_dict = guest.dict(exclude={"id"})
    update_dict = update.dict(exclude_unset=True)

    if "extra" in update_dict:
        if not validate_extra(update_dict["extra"], party_obj):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

        # overwrite allowed extra, but keep those that are not allowed/user-modifiable
        update_dict["extra"].update(
            {
                k: v
                for (k, v) in guest_dict["extra"].items()
                if k not in party_obj.allowed_extra
            }
        )

    guest_dict.update(update_dict)

    await db[party].replace_one({"_id": guest.id}, guest_dict)
    return await db[party].find_one({"_id": guest.id})


class PartyStatus(BaseModel):
    definitely_coming: int
    maybe_coming: int


@app.get(
    "/{party}/{token}/status",
    response_model=PartyStatus,
    responses=error_responses(401),
    tags=["guests"],
)
async def get_party_status(party: str, _=Depends(find_guest)):
    definitely_coming = await db[party].count_documents({"coming": "yes"})
    maybe_coming = await db[party].count_documents({"coming": "maybe"})

    return {
        "definitely_coming": definitely_coming,
        "maybe_coming": maybe_coming,
    }


# Admin methods


async def auth_admin(admin_token: str):
    if admin_token != settings.admin_token.get_secret_value():
        raise HTTPException(status.HTTP_401_UNAUTHORIZED)


@app.get(
    "/{admin_token}",
    response_model=list[Party],
    responses=error_responses(401),
    tags=["admin"],
)
async def list_parties(_=Depends(auth_admin)):
    return await meta["parties"].find().to_list(None)


class PartyCreate(BaseModel):
    name: str
    allowed_extra: list[str] = []


@app.post(
    "/{admin_token}",
    response_model=Party,
    status_code=status.HTTP_201_CREATED,
    responses=error_responses(400, 401),
    tags=["admin"],
)
async def create_party(party: PartyCreate, _=Depends(auth_admin)):
    try:
        await db.create_collection(party.name)
    except pymongo.errors.CollectionInvalid:
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST, f"Party {party.name!r} already exists"
        )

    party_dict = party.dict()
    party_dict.update({"created": datetime.now()})

    inserted = await meta["parties"].insert_one(party_dict)
    return await meta["parties"].find_one({"_id": inserted.inserted_id})


@app.delete(
    "/{admin_token}/{party}",
    status_code=status.HTTP_204_NO_CONTENT,
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def delete_party(party: str, _=Depends(auth_admin)):
    deleted = await meta["parties"].delete_one({"name": party})
    if deleted.deleted_count < 1:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    await db.drop_collection(party)


@app.get(
    "/{admin_token}/{party}",
    response_model=list[DBGuest],
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def list_guests(party: str, _=Depends(auth_admin)):
    if not await meta["parties"].find_one({"name": party}):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return await db[party].find().to_list(None)


class GuestCreate(BaseModel):
    token: str
    name: str
    coming: Coming | None
    grammatical_gender: GrammaticalGender
    extra: dict[str, str] = dict()


@app.post(
    "/{admin_token}/{party}",
    response_model=DBGuest,
    status_code=status.HTTP_201_CREATED,
    responses=error_responses(400, 401, 404),
    tags=["admin"],
)
async def create_new_guest(party: str, new_guest: GuestCreate, _=Depends(auth_admin)):
    await find_party(party)

    existing = await db[party].find_one({"token": new_guest.token})
    if existing:
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST, f"Token {new_guest.token!r} is already in use"
        )

    insert_result = await db[party].insert_one(new_guest.dict())
    inserted = await db[party].find_one({"_id": insert_result.inserted_id})
    return inserted


@app.get(
    "/{admin_token}/{party}/userAllowedExtra",
    response_model=list[str],
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def get_allowed_extra_keys(party: str, _=Depends(auth_admin)):
    party_obj = await find_party(party)
    return party_obj.allowed_extra


@app.patch(
    "/{admin_token}/{party}/userAllowedExtra",
    response_model=Party,
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def modify_allowed_extra_keys(party: str, keys: list[str], _=Depends(auth_admin)):
    party_obj = await find_party(party)
    party_dict = party_obj.dict(exclude={"id"})
    party_dict["allowed_extra"] = keys
    await meta["parties"].replace_one({"_id": party_obj.id}, party_dict)
    return await meta["parties"].find_one({"_id": party_obj.id})


class GuestModify(BaseModel):
    token: str | None
    name: str | None
    coming: Coming | None
    grammatical_gender: GrammaticalGender | None
    extra: dict[str, str] | None


@app.patch(
    "/{admin_token}/{party}/{id}",
    response_model=DBGuest,
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def modify_guest(
    party: str, id: PyObjectId, modified_guest: GuestModify, _=Depends(auth_admin)
):
    await find_party(party)

    existing = await db[party].find_one({"_id": id})
    if not existing:
        raise HTTPException(status.HTTP_404_NOT_FOUND)

    existing.update(modified_guest.dict(exclude={"id"}, exclude_unset=True))

    await db[party].replace_one({"_id": existing["_id"]}, existing)
    return await db[party].find_one({"_id": existing["_id"]})


@app.delete(
    "/{admin_token}/{party}/{id}",
    status_code=status.HTTP_204_NO_CONTENT,
    responses=error_responses(401, 404),
    tags=["admin"],
)
async def delete_guest(party: str, id: PyObjectId, _=Depends(auth_admin)):
    deleted = await db[party].delete_one({"_id": id})
    if deleted.deleted_count < 1:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)