fastapi_middleware.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_diagnostics/fastapi_middleware.py
 
"""
FastAPI / ASGI Diagnostics Middleware
======================================
 
Purpose
-------
Pure ASGI middleware that logs every HTTP request and response through
``_dbg()``, providing structured per-request diagnostics: HTTP method,
path, status code, client IP, and elapsed wall-clock time.
 
Responsibilities
----------------
- Implement ``DiagnosticsMiddleware`` as a pure ASGI callable that wraps
  any ASGI application without requiring ``starlette.BaseHTTPMiddleware``.
- Log request entry at L2 (method, path, client IP).
- Log User-Agent header at L3 and all request/response headers at L4.
- Capture HTTP status code from the ``http.response.start`` ASGI event via
  a lightweight ``send`` wrapper; no request or response bodies are buffered.
- Measure and report wall-clock elapsed time in milliseconds.
- Pass non-HTTP scopes (WebSocket, lifespan) through unchanged.
- Log application-level exceptions at L2 before re-raising them.
- Route all output through ``_dbg()`` using the configurable domain
  (default: ``"HTTP"``).
 
Diagnostics
-----------
Domain: HTTP
Levels:
    L2 — lifecycle (request start, response end, application errors)
    L3 — semantic details (User-Agent, 4xx/5xx status class)
    L4 — deep tracing (all request and response headers)
 
Contracts
---------
- Must never suppress exceptions raised by the wrapped application;
  errors are logged at L2 and then re-raised unchanged.
- Must not buffer or modify request or response bodies.
- Must pass non-HTTP ASGI scopes (WebSocket, lifespan) through unchanged.
- Does not require FastAPI at import time; works with any ASGI framework.
  Install the ``[fastapi]`` optional extra for FastAPI framework utilities::
 
      pip install oc_diagnostics[fastapi]
 
- The ``HTTP`` domain must be configured in the project's
  ``diagnostics/debug_config.json`` with a non-zero level for log output
  to appear (``DEBUG_LEVELS.get("HTTP", 0)`` defaults to disabled).
"""
 
import time
 
import oc_diagnostics.unified_debug as _ud
from oc_diagnostics.unified_debug import _dbg
 
 
# ============================================================
# PUBLIC MIDDLEWARE CLASS
# ============================================================
 
class DiagnosticsMiddleware:
    """
    Pure ASGI middleware that logs HTTP requests and responses via ``_dbg()``.
 
    Parameters
    ----------
    app : ASGI callable
        The next ASGI application in the middleware stack.
    domain : str, optional
        Debug domain used for all log output.  Defaults to ``"HTTP"``.
        Must be configured in ``diagnostics/debug_config.json`` with a
        non-zero level; unknown domains are silently filtered by ``_dbg()``.
 
    Usage
    -----
    Add to a FastAPI application **after** all routes are registered::
 
        from fastapi import FastAPI
        from oc_diagnostics import DiagnosticsMiddleware
 
        app = FastAPI()
        app.add_middleware(DiagnosticsMiddleware)
 
    With a custom domain::
 
        app.add_middleware(DiagnosticsMiddleware, domain="API")
 
    Enable output by adding the ``HTTP`` domain to ``diagnostics/debug_config.json``::
 
        "HTTP": { "level": 2, "color": "blue" }
 
    Level guide for this middleware:
        L2 — request/response lifecycle lines (always recommended)
        L3 — User-Agent, 4xx/5xx status class commentary
        L4 — full request and response header dumps
    """
 
    def __init__(self, app, domain: str = "HTTP"):
        self._app = app
        self._domain = domain
        _dbg(
            self._domain,
            "init",
            f"DiagnosticsMiddleware registered (domain={domain!r})",
            level=2,
        )
 
    async def __call__(self, scope, receive, send):
        """
        ASGI entry point.
 
        HTTP scopes are intercepted and logged.  All other scope types
        (``"websocket"``, ``"lifespan"``) are passed through unchanged.
        """
        # --------------------------------------------------------
        # Non-HTTP scopes: pass through without any logging
        # --------------------------------------------------------
        if scope["type"] != "http":
            await self._app(scope, receive, send)
            return
 
        # --------------------------------------------------------
        # Extract request metadata
        # --------------------------------------------------------
        func = "request"
        method = scope.get("method", "?")
        path = scope.get("path", "/")
        query = scope.get("query_string", b"").decode("utf-8", errors="replace")
        full_path = f"{path}?{query}" if query else path
        client = scope.get("client")
        client_ip = f"{client[0]}:{client[1]}" if client else "unknown"
 
        # L2 — request start
        _dbg(
            self._domain,
            func,
            f"START {method} {full_path}  client={client_ip}",
            level=2,
        )
 
        # L3 — User-Agent header
        try:
            headers_raw = scope.get("headers", [])
            header_map = {k.lower(): v for k, v in headers_raw}
            user_agent = header_map.get(b"user-agent", b"").decode("utf-8", errors="replace")
            if user_agent:
                _dbg(self._domain, func, f"User-Agent: {user_agent}", level=3)
        except Exception:
            pass
 
        # L4 — all request headers (FS-17: sensitive headers are redacted)
        try:
            for key, value in scope.get("headers", []):
                header_name = key.decode("utf-8", errors="replace").lower()
                display_value = (
                    "[REDACTED]"
                    if header_name in _ud.HTTP_REDACT_HEADERS
                    else value.decode("utf-8", errors="replace")
                )
                _dbg(
                    self._domain,
                    func,
                    f"req-header  {header_name}: {display_value}",
                    level=4,
                )
        except Exception:
            pass
 
        # --------------------------------------------------------
        # Wrap send() to capture status code and response headers
        # --------------------------------------------------------
        start = time.perf_counter()
        status_holder = [None]
 
        async def _send_with_capture(message):
            if message["type"] == "http.response.start":
                status_holder[0] = message.get("status")
                # L4 — response headers (FS-17: sensitive headers are redacted)
                try:
                    for key, value in message.get("headers", []):
                        header_name = key.decode("utf-8", errors="replace").lower()
                        display_value = (
                            "[REDACTED]"
                            if header_name in _ud.HTTP_REDACT_HEADERS
                            else value.decode("utf-8", errors="replace")
                        )
                        _dbg(
                            self._domain,
                            func,
                            f"resp-header {header_name}: {display_value}",
                            level=4,
                        )
                except Exception:
                    pass
            await send(message)
 
        # --------------------------------------------------------
        # Run wrapped application; log errors and re-raise
        # --------------------------------------------------------
        try:
            await self._app(scope, receive, _send_with_capture)
        except Exception as exc:
            elapsed_ms = (time.perf_counter() - start) * 1000
            _dbg(
                self._domain,
                func,
                f"ERROR {method} {full_path}  [{elapsed_ms:.1f}ms]  "
                f"{type(exc).__name__}: {exc}",
                level=2,
            )
            raise
 
        # --------------------------------------------------------
        # L2 — request end
        # --------------------------------------------------------
        elapsed_ms = (time.perf_counter() - start) * 1000
        status = status_holder[0] if status_holder[0] is not None else "?"
        _dbg(
            self._domain,
            func,
            f"END   {method} {full_path}  -> {status}  [{elapsed_ms:.1f}ms]",
            level=2,
        )
 
        # L3 — status class commentary
        try:
            if isinstance(status, int):
                if status >= 500:
                    _dbg(self._domain, func, f"5xx server error: {status}", level=3)
                elif status >= 400:
                    _dbg(self._domain, func, f"4xx client error: {status}", level=3)
        except Exception:
            pass