made fastapi tools

python/api/dependencies.py

@@ -1,16 +0,0 @@
-"""FastAPI dependencies."""
-
-from collections.abc import Iterator
-from typing import Annotated
-
-from fastapi import Depends, Request
-from sqlalchemy.orm import Session
-
-
-def get_db(request: Request) -> Iterator[Session]:
-    """Get database session from app state."""
-    with Session(request.app.state.engine) as session:
-        yield session
-
-
-DbSession = Annotated[Session, Depends(get_db)]

python/api/main.py

@@ -9,9 +9,9 @@ import typer
 import uvicorn
 from fastapi import FastAPI
 
-from python.api.middleware import ZstdMiddleware
 from python.api.routers import contact_router, views_router
 from python.common import configure_logger
+from python.fastapi_tools import ZstdMiddleware
 from python.orm.common import get_postgres_engine
 
 logger = logging.getLogger(__name__)

python/api/middleware.py

@@ -1,49 +0,0 @@
-"""Middleware for the FastAPI application."""
-
-from compression import zstd
-from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
-from starlette.requests import Request
-from starlette.responses import Response
-
-MINIMUM_RESPONSE_SIZE = 500
-
-
-class ZstdMiddleware(BaseHTTPMiddleware):
-    """Middleware that compresses responses with zstd when the client supports it."""
-
-    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
-        """Compress the response with zstd if the client accepts it."""
-        accepted_encodings = request.headers.get("accept-encoding", "")
-        if "zstd" not in accepted_encodings:
-            return await call_next(request)
-
-        response = await call_next(request)
-
-        if response.headers.get("content-encoding") or "text/event-stream" in response.headers.get("content-type", ""):
-            return response
-
-        body = b""
-        async for chunk in response.body_iterator:
-            body += chunk if isinstance(chunk, bytes) else chunk.encode()
-
-        if len(body) < MINIMUM_RESPONSE_SIZE:
-            return Response(
-                content=body,
-                status_code=response.status_code,
-                headers=dict(response.headers),
-                media_type=response.media_type,
-            )
-
-        compressed = zstd.compress(body)
-
-        headers = dict(response.headers)
-        headers["content-encoding"] = "zstd"
-        headers["content-length"] = str(len(compressed))
-        headers.pop("transfer-encoding", None)
-
-        return Response(
-            content=compressed,
-            status_code=response.status_code,
-            headers=headers,
-            media_type=response.media_type,
-        )

python/api/routers/contact.py

@@ -9,7 +9,7 @@ from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import selectinload
 
-from python.api.dependencies import DbSession
+from python.fastapi_tools.db import DbSession
 from python.orm.richie.contact import Contact, ContactRelationship, Need, RelationshipType
 
 TEMPLATES_DIR = Path(__file__).parent.parent / "templates"

python/api/routers/views.py

@@ -9,7 +9,7 @@ from fastapi.templating import Jinja2Templates
 from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
 
-from python.api.dependencies import DbSession
+from python.fastapi_tools.db import DbSession
 from python.orm.richie.contact import Contact, ContactRelationship, Need, RelationshipType
 
 TEMPLATES_DIR = Path(__file__).parent.parent / "templates"