Spaces:
Running
Running
| """Middleware for the API.""" | |
| from fastapi import Request | |
| from starlette.middleware.base import BaseHTTPMiddleware | |
| from starlette.responses import Response | |
| import time | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| class RequestLoggingMiddleware(BaseHTTPMiddleware): | |
| """Middleware to log all requests and their processing time.""" | |
| async def dispatch(self, request: Request, call_next) -> Response: | |
| """Log request and response information.""" | |
| start_time = time.time() | |
| # Log request | |
| logger.info(f"Request: {request.method} {request.url.path}") | |
| # Process request | |
| response = await call_next(request) | |
| # Calculate processing time | |
| process_time = time.time() - start_time | |
| # Log response | |
| logger.info( | |
| f"Response: {request.method} {request.url.path} " | |
| f"Status: {response.status_code} " | |
| f"Duration: {process_time:.3f}s" | |
| ) | |
| # Add custom header with processing time | |
| response.headers["X-Process-Time"] = str(process_time) | |
| return response | |
| class SecurityHeadersMiddleware(BaseHTTPMiddleware): | |
| """Middleware to add security headers to responses.""" | |
| async def dispatch(self, request: Request, call_next) -> Response: | |
| """Add security headers to response.""" | |
| response = await call_next(request) | |
| # Add security headers | |
| response.headers["X-Content-Type-Options"] = "nosniff" | |
| response.headers["X-Frame-Options"] = "DENY" | |
| response.headers["X-XSS-Protection"] = "1; mode=block" | |
| response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" | |
| return response | |