| """Request ID + structured access logging for production operations.""" |
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import time |
| import uuid |
| from typing import Callable |
|
|
| from starlette.middleware.base import BaseHTTPMiddleware |
| from starlette.requests import Request |
| from starlette.responses import Response |
|
|
| logger = logging.getLogger("cepheus.access") |
|
|
|
|
| class RequestContextMiddleware(BaseHTTPMiddleware): |
| async def dispatch(self, request: Request, call_next: Callable) -> Response: |
| request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4()) |
| request.state.request_id = request_id |
| start = time.perf_counter() |
| status_code = 500 |
| try: |
| response = await call_next(request) |
| status_code = response.status_code |
| response.headers["X-Request-ID"] = request_id |
| return response |
| finally: |
| duration_ms = round((time.perf_counter() - start) * 1000, 2) |
| log_entry = { |
| "request_id": request_id, |
| "method": request.method, |
| "path": request.url.path, |
| "status": status_code, |
| "duration_ms": duration_ms, |
| "client": request.client.host if request.client else None, |
| } |
| if status_code >= 500: |
| logger.error(json.dumps(log_entry)) |
| elif status_code >= 400: |
| logger.warning(json.dumps(log_entry)) |
| else: |
| logger.info(json.dumps(log_entry)) |
|
|
