79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
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
|
|
|
|
|
|
# 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",
|
|
)
|