presentation.routers.api.middleware.trace_middleware

src.presentation.routers.api.middleware.trace_middleware

Trace middleware to inject a trace_id per request.

• Adds X-Trace-Id response header
• Exposes get_trace_id() helper for logging calls outside request handlers

Classes

TraceMiddleware

Bases: BaseHTTPMiddleware

Starlette middleware that injects a trace ID into each request context.

Source code in src/presentation/routers/api/middleware/trace_middleware.py

class TraceMiddleware(BaseHTTPMiddleware):
    """Starlette middleware that injects a trace ID into each request context."""

    async def dispatch(
        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        """Intercept a request to set and propagate a trace ID.

        Args:
            request (Request): Incoming request.
            call_next (Callable[[Request], Awaitable[Response]]): Next handler.

        Returns:
            Response: Response with X-Trace-Id header added.
        """
        trace_id = request.headers.get("X-Trace-Id") or str(uuid7())
        trace_id_context.set(trace_id)
        try:
            response = await call_next(request)
            response.headers["X-Trace-Id"] = trace_id
            return response
        finally:
            # Clear context after request to prevent leakage
            trace_id_context.set(None)

Functions

dispatch async

dispatch(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response

Intercept a request to set and propagate a trace ID.

Parameters:

Name	Type	Description	Default
request	Request	Incoming request.	required
call_next	Callable[[Request], Awaitable[Response]]	Next handler.	required

Returns:

Name	Type	Description
Response	Response	Response with X-Trace-Id header added.

Source code in src/presentation/routers/api/middleware/trace_middleware.py

async def dispatch(
    self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    """Intercept a request to set and propagate a trace ID.

    Args:
        request (Request): Incoming request.
        call_next (Callable[[Request], Awaitable[Response]]): Next handler.

    Returns:
        Response: Response with X-Trace-Id header added.
    """
    trace_id = request.headers.get("X-Trace-Id") or str(uuid7())
    trace_id_context.set(trace_id)
    try:
        response = await call_next(request)
        response.headers["X-Trace-Id"] = trace_id
        return response
    finally:
        # Clear context after request to prevent leakage
        trace_id_context.set(None)

Functions

get_trace_id

get_trace_id() -> str | None

Return the current trace ID.

Returns None when called outside of request context. Use this in logging calls to automatically include trace_id.

Returns:

Type	Description
str | None	str | None: The current request trace ID, or None if no active request.

Source code in src/presentation/routers/api/middleware/trace_middleware.py

def get_trace_id() -> str | None:
    """Return the current trace ID.

    Returns None when called outside of request context.
    Use this in logging calls to automatically include trace_id.

    Returns:
        str | None: The current request trace ID, or None if no active request.
    """
    return trace_id_context.get()