| """ |
| API middleware for request logging, rate limiting, and error handling. |
| """ |
|
|
| from fastapi import Request |
| from fastapi.responses import JSONResponse |
| from starlette.middleware.base import BaseHTTPMiddleware |
| from loguru import logger |
| import time |
| from collections import defaultdict, deque |
|
|
|
|
| class RequestLoggingMiddleware(BaseHTTPMiddleware): |
| """Logs all incoming requests with timing information.""" |
|
|
| async def dispatch(self, request: Request, call_next): |
| start_time = time.time() |
| path = request.url.path |
| method = request.method |
|
|
| logger.info(f"→ {method} {path}") |
|
|
| try: |
| response = await call_next(request) |
| except Exception as e: |
| logger.error(f"✗ {method} {path} - Error: {e}") |
| raise |
|
|
| elapsed = (time.time() - start_time) * 1000 |
| logger.info(f"← {method} {path} - {response.status_code} ({elapsed:.1f}ms)") |
|
|
| return response |
|
|
|
|
| class RateLimitMiddleware(BaseHTTPMiddleware): |
| """Simple in-memory rate limiting.""" |
|
|
| def __init__(self, app, max_requests_per_minute: int = 60): |
| super().__init__(app) |
| self.max_requests = max_requests_per_minute |
| self.window = 60 |
| |
| self.requests: dict = defaultdict(deque) |
|
|
| async def dispatch(self, request: Request, call_next): |
| |
| client_ip = request.client.host if request.client else "unknown" |
| now = time.time() |
|
|
| |
| timestamps = self.requests[client_ip] |
| while timestamps and timestamps[0] < now - self.window: |
| timestamps.popleft() |
|
|
| |
| if len(timestamps) >= self.max_requests: |
| logger.warning(f"Rate limited: {client_ip} ({len(timestamps)} requests in {self.window}s)") |
| return JSONResponse( |
| status_code=429, |
| content={"detail": "Rate limit exceeded. Please wait before making more requests."}, |
| ) |
|
|
| |
| timestamps.append(now) |
|
|
| response = await call_next(request) |
| return response |
|
|
