munch-ease-backend/api/deps.py

from __future__ import annotations

from typing import AsyncGenerator, Optional

import aiosqlite
from fastapi import Cookie, Depends, HTTPException, Request
from fastapi.responses import JSONResponse

import db
import persons
from common import ProblemDetails
from settings import settings
from users import repository as users_db
from users.models import User


# Dependency to create SQLite connection with PRAGMAs and per-request transaction
async def get_db() -> AsyncGenerator[aiosqlite.Connection, None]:
    sql_db = await db.connect(settings.database_path)
    # Connection-level configuration
    try:
        # Enable FK enforcement
        await sql_db.execute("PRAGMA foreign_keys=ON;")
        # Prefer WAL for better concurrency; ignore result
        async with sql_db.execute("PRAGMA journal_mode=WAL;") as _:
            await _.fetchone()
        # Reasonable durability/perf tradeoff
        await sql_db.execute("PRAGMA synchronous=NORMAL;")
        # Begin a transaction for the whole request
        await sql_db.execute("BEGIN;")

        try:
            yield sql_db
            await sql_db.commit()
        except Exception:
            await sql_db.rollback()
            raise
    finally:
        await sql_db.close()


async def cookie_person(
    user_id: int = Cookie(..., alias="user_id"),
    conn: aiosqlite.Connection = Depends(get_db),
) -> persons.Person:
    """Return the authenticated user from the user_id cookie or raise 401.

    When the cookie is missing, FastAPI will raise 422 (validation error).
    """
    person = await persons.get_by_id(conn, user_id)
    if not person:
        raise HTTPException(status_code=401, detail="Unauthorized")
    return person


async def cookie_person_optional(
    user_id: Optional[int] = Cookie(default=None, alias="user_id"),
    conn: aiosqlite.Connection = Depends(get_db),
) -> Optional[persons.Person]:
    """Return the authenticated user if cookie present; otherwise None.

    Use for endpoints that want to return 401 for missing auth themselves.
    """
    if user_id is None:
        return None
    person = await persons.get_by_id(conn, user_id)
    return person


def error_response(request: Optional[Request], status_code: int, message: str) -> JSONResponse:
    body = ProblemDetails(
        title=message,
        status=status_code,
        type=f"https://httpstatuses.com/{status_code}",
        instance=str(request.url) if request else None,
    )
    return JSONResponse(
        content=body.model_dump(by_alias=True),
        status_code=status_code,
        media_type="application/problem+json",
    )


async def get_current_user(request: Request, conn: aiosqlite.Connection = Depends(get_db)) -> User:
    """Temporary bearer token auth: expects Authorization: Bearer user-<id>.

    This is a stopgap until JWT is implemented. Returns 401 on failure.
    """
    auth = request.headers.get("Authorization")
    if not auth or not auth.lower().startswith("bearer "):
        raise HTTPException(status_code=401, detail="Unauthorized")
    token = auth.split(" ", 1)[1].strip()
    if not token.startswith("user-"):
        raise HTTPException(status_code=401, detail="Unauthorized")
    try:
        user_id = int(token.split("-", 1)[1])
    except Exception:
        raise HTTPException(status_code=401, detail="Unauthorized")
    # Lookup by id
    async with conn.execute(
        "SELECT id, email, display_name, profile_photo_url FROM User WHERE id = ?",
        (user_id,),
    ) as c:
        row = await c.fetchone()
        if not row:
            raise HTTPException(status_code=401, detail="Unauthorized")
        return User(id=int(row[0]), email=row[1], display_name=row[2], profile_photo_url=row[3])