munch-ease-backend/api/deps.py

from __future__ import annotations

from typing import Annotated, AsyncGenerator, Optional

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

import db
import persons
from common import ProblemDetails
from settings import settings


# 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: Optional[int] = Cookie(None, alias="user_id"),
    conn: aiosqlite.Connection = Depends(get_db),
) -> Optional[persons.Person]:
    if user_id is None:
        return None
    return await persons.get_by_id(conn, user_id)


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",
    )