File size: 4,332 Bytes
d5ef46f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
Logging Middleware for FastAPI
Automatically logs requests and responses using logfire when available.
Follows 2025 best practices for simple, automatic instrumentation.
"""
import time
from collections.abc import Callable
from typing import cast
from fastapi import Request, Response
from fastapi.routing import APIRoute
from starlette.middleware.base import BaseHTTPMiddleware
from ..config.logfire_config import LOGFIRE_AVAILABLE, get_logger, is_logfire_enabled
class LoggingMiddleware(BaseHTTPMiddleware):
"""
Middleware that automatically logs HTTP requests and responses.
Skips health check endpoints to reduce noise.
"""
SKIP_PATHS = {"/health", "/api/health", "/", "/docs", "/redoc", "/openapi.json"}
def __init__(self, app):
super().__init__(app)
self.logger = get_logger("middleware")
async def dispatch(self, request: Request, call_next: Callable) -> Response:
# Skip logging for certain paths
if request.url.path in self.SKIP_PATHS:
return cast(Response, await call_next(request))
# Record start time
start_time = time.time()
# Log the request
client_host = request.client.host if request.client is not None else "unknown"
self.logger.info(f"HTTP Request | method={request.method} | path={request.url.path} | client={client_host}")
try:
# Process the request
response = cast(Response, await call_next(request))
# Calculate duration
duration = time.time() - start_time
# Log the response
self.logger.info(
f"HTTP Response | method={request.method} | path={request.url.path} | status_code={response.status_code} | duration_ms={round(duration * 1000, 2)}"
)
return response
except Exception as e:
# Log errors
duration = time.time() - start_time
self.logger.error(
f"HTTP Error | method={request.method} | path={request.url.path} | error={str(e)} | duration_ms={round(duration * 1000, 2)}"
)
raise
def instrument_fastapi(app):
"""
Instrument a FastAPI app with automatic logging.
This is the recommended approach for 2025 - let logfire handle the complexity.
"""
logger = get_logger("instrumentation")
if is_logfire_enabled() and LOGFIRE_AVAILABLE:
try:
# Import logfire for instrumentation only when enabled
import logfire
# Use logfire's built-in FastAPI instrumentation
logfire.instrument_fastapi(app)
logger.info("FastAPI instrumented with logfire")
except Exception as e:
logger.error(f"Failed to instrument FastAPI with logfire: {e}")
# Fall back to our custom middleware
app.add_middleware(LoggingMiddleware)
else:
# Use our custom middleware for basic logging
app.add_middleware(LoggingMiddleware)
logger.info("FastAPI instrumented with custom logging middleware")
class LoggingRoute(APIRoute):
"""
Custom APIRoute that logs endpoint execution time.
This provides more granular logging than middleware alone.
"""
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
logger = get_logger("endpoint")
async def custom_route_handler(request: Request) -> Response:
start_time = time.time()
# Get endpoint info
endpoint_name = self.endpoint.__name__ if self.endpoint is not None else "unknown"
try:
response = await original_route_handler(request)
duration = time.time() - start_time
# Log successful endpoint execution
logger.info(f"Endpoint: {endpoint_name} | duration_ms={round(duration * 1000, 2)} | status=success")
return response
except Exception as e:
duration = time.time() - start_time
# Log endpoint error
logger.error(
f"Endpoint: {endpoint_name} | duration_ms={round(duration * 1000, 2)} | status=error | error={str(e)}"
)
raise
return custom_route_handler
|