MCP-OPENAPI-DOCX

"""Logging middleware. This module provides request/response logging middleware. """ import time import uuid from collections.abc import Callable import structlog from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware logger = structlog.get_logger(__name__) class LoggingMiddleware(BaseHTTPMiddleware): """Middleware for request/response logging. This middleware logs incoming requests and outgoing responses with timing information. """ async def dispatch( self, request: Request, call_next: Callable, ) -> Response: """Process the request and log details. Args: request: Incoming request. call_next: Next middleware/handler. Returns: Response from the next handler. """ # Generate request ID request_id = str(uuid.uuid4()) request.state.request_id = request_id # Record start time start_time = time.time() # Get client info client_ip = request.client.host if request.client else "unknown" user_agent = request.headers.get("User-Agent", "unknown") # Log request logger.info( "request_started", request_id=request_id, method=request.method, path=request.url.path, client_ip=client_ip, user_agent=user_agent, ) # Process request try: response = await call_next(request) except Exception as e: # Log error duration = time.time() - start_time logger.error( "request_failed", request_id=request_id, method=request.method, path=request.url.path, duration=duration, error=str(e), ) raise # Calculate duration duration = time.time() - start_time # Add request ID header response.headers["X-Request-ID"] = request_id # Log response logger.info( "request_completed", request_id=request_id, method=request.method, path=request.url.path, status_code=response.status_code, duration=duration, ) return response