Middleware Stack

ZodiacCore provides a standard stack of Pure ASGI middlewares for request tracing, latency monitoring, and access logging. They handle HTTP and WebSocket; lifespan scope is passed through unchanged.

1. Core Middlewares

Trace ID Middleware

The TraceIDMiddleware is the entry point for observability. It:

1. Reads: Looks for an X-Request-ID header in the incoming request (HTTP) or WebSocket upgrade request.
2. Generates: If missing or invalid (not 36 characters), it generates a fresh UUID.
3. Persists: Sets the ID in the request context (via zodiac_core.context) for the duration of the request or WebSocket connection, then resets it.
4. Responds: For HTTP, attaches the same ID to the response headers for frontend tracking.

Access Log Middleware

The AccessLogMiddleware records every HTTP request and WebSocket connection. It logs:

• HTTP: Method, path, status code, and processing latency (ms). Trace ID is picked up from context when present.
• WebSocket: Path and latency with a fixed status 101 (Switching Protocols). Trace ID is available in context for the connection lifetime.
• Lifespan: Not logged; scope is passed through.

---

2. Usage & Order

The simplest way to use these is via register_middleware.

Middleware Order

ZodiacCore adds middlewares in a specific order to ensure that the Trace ID is generated before the Access Log tries to record it.

from fastapi import FastAPI
from zodiac_core.middleware import register_middleware

app = FastAPI()

# Registers both TraceID and AccessLog middlewares in the correct order
register_middleware(app)

---

3. Customizing Trace ID Generation

If you want to use a custom header name or a different ID generator (e.g., K-Sorted IDs), you can add the middleware manually:

from zodiac_core.middleware import TraceIDMiddleware

app.add_middleware(
    TraceIDMiddleware,
    header_name="X-Correlation-ID",
    generator=lambda: "my-custom-id-123"
)

---

4. API Reference

Middleware Utilities

Middleware stack: Trace ID and Access Log.

Implemented as Pure ASGI middleware (no BaseHTTPMiddleware).

Scope types ("http", "websocket", "lifespan") follow the ASGI spec: - https://asgi.readthedocs.io/en/stable/specs/www.html (http, websocket) - https://asgi.readthedocs.io/en/stable/specs/lifespan.html (lifespan)

TraceIDMiddleware

Request ID middleware (Pure ASGI).

1. Extracts or generates X-Request-ID.
2. Sets it in a ContextVar (zodiac_core.context).
3. Appends it to the response headers.

Compatible with: app.add_middleware(TraceIDMiddleware) and app.add_middleware(TraceIDMiddleware, header_name="...", generator=...).

Source code in zodiac_core/middleware.py

class TraceIDMiddleware:
    """
    Request ID middleware (Pure ASGI).

    1. Extracts or generates X-Request-ID.
    2. Sets it in a ContextVar (zodiac_core.context).
    3. Appends it to the response headers.

    Compatible with: app.add_middleware(TraceIDMiddleware) and
    app.add_middleware(TraceIDMiddleware, header_name="...", generator=...).
    """

    def __init__(
        self,
        app: ASGIApp,
        header_name: str = "X-Request-ID",
        generator: Callable[[], str] | None = None,
    ) -> None:
        self.app = app
        self.header_name = header_name
        self.generator = generator or default_id_generator

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # ASGI scope["type"]: "http" | "websocket" | "lifespan" (see module docstring links)
        if scope["type"] == "http":
            headers = MutableHeaders(scope=scope)
            header_value = headers.get(self.header_name)
            request_id = self.generator() if header_value is None or len(header_value) != 36 else header_value
            with request_id_scope(request_id):

                async def send_wrapper(message: Message) -> None:
                    if message["type"] == "http.response.start":
                        out_headers = MutableHeaders(scope=message)
                        out_headers.append(self.header_name, request_id)
                    await send(message)

                await self.app(scope, receive, send_wrapper)
            return
        if scope["type"] == "websocket":
            headers = MutableHeaders(scope=scope)
            header_value = headers.get(self.header_name)
            request_id = self.generator() if header_value is None or len(header_value) != 36 else header_value
            with request_id_scope(request_id):
                await self.app(scope, receive, send)
            return
        await self.app(scope, receive, send)

AccessLogMiddleware

Access log middleware (Pure ASGI).

Logs method, path, status code, and latency. Uses loguru; request_id appears in logs when TraceIDMiddleware is used (context).

Compatible with: app.add_middleware(AccessLogMiddleware).

Source code in zodiac_core/middleware.py

class AccessLogMiddleware:
    """
    Access log middleware (Pure ASGI).

    Logs method, path, status code, and latency. Uses loguru; request_id
    appears in logs when TraceIDMiddleware is used (context).

    Compatible with: app.add_middleware(AccessLogMiddleware).
    """

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # ASGI scope["type"]: "http" | "websocket" | "lifespan" (see module docstring links)
        if scope["type"] == "http":
            start_time = time.perf_counter()
            status_code = 500

            async def send_wrapper(message: Message) -> None:
                nonlocal status_code
                if message["type"] == "http.response.start":
                    status_code = message.get("status", 500)
                await send(message)

            try:
                await self.app(scope, receive, send_wrapper)
            finally:
                process_time = (time.perf_counter() - start_time) * 1000
                logger.info(
                    "{method} {path} - {status_code} - {latency:.2f}ms",
                    method=scope.get("method", "GET"),
                    path=scope.get("path", "/"),
                    status_code=status_code,
                    latency=process_time,
                )
            return
        if scope["type"] == "websocket":
            start_time = time.perf_counter()
            try:
                await self.app(scope, receive, send)
            finally:
                process_time = (time.perf_counter() - start_time) * 1000
                logger.info(
                    "WEBSOCKET {path} - 101 - {latency:.2f}ms",
                    path=scope.get("path", "/"),
                    latency=process_time,
                )
            return
        await self.app(scope, receive, send)

register_middleware(app)

Register TraceID and AccessLog middlewares in the correct order.

Order: TraceID (outer) then AccessLog (inner), so the access log can include the request_id from context.

Source code in zodiac_core/middleware.py

def register_middleware(app: ASGIApp) -> None:
    """
    Register TraceID and AccessLog middlewares in the correct order.

    Order: TraceID (outer) then AccessLog (inner), so the access log
    can include the request_id from context.
    """
    app.add_middleware(AccessLogMiddleware)
    app.add_middleware(TraceIDMiddleware)