Skip to content

Middleware Stack

ZodiacCore provides a standard stack of Pure ASGI middlewares for request tracing, service log attribution, 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.

Service Name Middleware

The ServiceNameMiddleware stores the current service name in request context so logs can attribute entries to the correct app or mounted service. It is most useful when multiple apps share one process-wide logger sink, or when you want every request log to carry an explicit service label.

Use it through register_middleware(service_name=...) instead of adding it manually in normal applications.

Access Log Middleware

The AccessLogMiddleware records HTTP requests and WebSocket connections. 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.

Use exclude_paths to skip noisy HTTP access logs such as health checks. Paths are matched exactly against scope["path"], without query strings. For example, "/api/v1/health" matches only that path.

2. Usage & Order

The simplest way to use these is via register_middleware. Call it once per app and pass all options in the same call.

Middleware Order

ZodiacCore adds middlewares in a specific order: Trace ID is set first, Service Name is scoped next when configured, and Access Log records after the downstream app finishes.

from fastapi import FastAPI
from zodiac_core.middleware import register_middleware

app = FastAPI()

register_middleware(
    app,
    service_name="inventory-service",
    exclude_paths=[
        "/api/v1/health",
    ],
)

For multi‑app deployments with app.mount(), see the Sub Applications guide.

exclude_paths affects HTTP access logs only. WebSocket access logs are still recorded.

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 
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
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).

Use exclude_paths to skip noisy endpoints such as health checks. Paths are matched exactly against scope["path"].

Source code in zodiac_core/middleware.py 
 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
130
131
132
133
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).

    Use ``exclude_paths`` to skip noisy endpoints such as health checks.
    Paths are matched exactly against ``scope["path"]``.
    """

    def __init__(
        self,
        app: ASGIApp,
        exclude_paths: Sequence[str] | None = None,
    ) -> None:
        self.app = app
        self._exclude_paths = frozenset(exclude_paths or ())

    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
            path = scope.get("path", "/")

            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:
                if path not in self._exclude_paths:
                    process_time = (time.perf_counter() - start_time) * 1000
                    logger.info(
                        "{method} {path} - {status_code} - {latency:.2f}ms",
                        method=scope.get("method", "GET"),
                        path=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)

ServiceNameMiddleware

Service name middleware (Pure ASGI).

Sets a per-request service name in context so mounted apps can keep app-local log attribution while sharing the process-wide logger sinks.

Source code in zodiac_core/middleware.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class ServiceNameMiddleware:
    """
    Service name middleware (Pure ASGI).

    Sets a per-request service name in context so mounted apps can keep
    app-local log attribution while sharing the process-wide logger sinks.
    """

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

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] in {"http", "websocket"}:
            with service_name_scope(self.service_name):
                await self.app(scope, receive, send)
            return
        await self.app(scope, receive, send)

register_middleware(app, service_name=None, *, exclude_paths=None)

Register TraceID and AccessLog middlewares in the correct order.

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

Parameters:

Name Type Description Default
exclude_paths Sequence[str] | None

Optional exact paths to skip in access logs. Matches against scope["path"].

 None
Source code in zodiac_core/middleware.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def register_middleware(
    app: ASGIApp,
    service_name: str | None = None,
    *,
    exclude_paths: Sequence[str] | None = None,
) -> None:
    """
    Register TraceID and AccessLog middlewares in the correct order.

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

    Args:
        exclude_paths: Optional exact paths to skip in access logs. Matches
            against ``scope["path"]``.
    """
    app.add_middleware(
        AccessLogMiddleware,
        exclude_paths=exclude_paths,
    )
    if service_name is not None:
        app.add_middleware(ServiceNameMiddleware, service_name=service_name)
    app.add_middleware(TraceIDMiddleware)