"""Access logging middleware for FastAPI.""" import time from typing import Callable from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware from sqlalchemy import insert from database import async_session_factory from models.access_log import AccessLog class AccessLogMiddleware(BaseHTTPMiddleware): """Middleware that logs every API request to the access_logs table.""" # Paths to exclude from logging (health checks, static files) EXCLUDED_PATHS = {"/api/health", "/docs", "/openapi.json", "/redoc"} async def dispatch(self, request: Request, call_next: Callable) -> Response: # Skip excluded paths if request.url.path in self.EXCLUDED_PATHS: return await call_next(request) start_time = time.time() response = await call_next(request) duration_ms = (time.time() - start_time) * 1000 # Extract user info from request state (set by auth middleware) user_id = getattr(request.state, "user_id", None) # Log asynchronously (don't block response) try: async with async_session_factory() as session: await session.execute( insert(AccessLog).values( user_id=user_id, action=f"{request.method} {request.url.path}", details={ "method": request.method, "path": request.url.path, "query": str(request.query_params), "status_code": response.status_code, "duration_ms": round(duration_ms, 2), }, ip_address=request.client.host if request.client else None, user_agent=request.headers.get("user-agent", "")[:500], ) ) await session.commit() except Exception: # Don't fail the request if logging fails pass return response