42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Annotated, Optional, AsyncGenerator
|
||
|
|
|
||
|
|
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
|
||
|
|
async def get_db() -> AsyncGenerator[aiosqlite.Connection, None]:
|
||
|
|
sql_db = await db.connect(settings.database_path)
|
||
|
|
try:
|
||
|
|
yield sql_db
|
||
|
|
finally:
|
||
|
|
await sql_db.close()
|
||
|
|
|
||
|
|
|
||
|
|
async def cookie_person(
|
||
|
|
user_id: Annotated[int, Cookie(alias="user_id")], conn: aiosqlite.Connection = Depends(get_db)
|
||
|
|
) -> Optional[persons.Person]:
|
||
|
|
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",
|
||
|
|
)
|