Tracing & HTTP Clients

ZodiacCore provides built-in support for Distributed Tracing. It ensures that a single Request ID (Trace ID) flows through your entire ecosystem: from the incoming request, through your logs, and out to downstream microservices via HTTP clients.

1. How it Works

1. Extraction: TraceIDMiddleware catches X-Request-ID from incoming HTTP or WebSocket upgrade headers (or generates a new one).
2. Storage: The ID is stored in a thread-safe ContextVar (managed by zodiac_core.context) and reset when the request or connection ends (via request_id_scope).
3. Observation: Logging utilities automatically pick up this ID from the context.
4. Propagation: ZodiacClient automatically injects this ID into outgoing HTTP requests.

---

2. Distributed Tracing (HTTP Clients)

When calling other services, use ZodiacClient (Async) or ZodiacSyncClient (Sync). They are thin wrappers around httpx that automatically handle Trace ID propagation.

Async Usage (Recommended)

from zodiac_core.http import ZodiacClient

async def call_downstream():
    async with ZodiacClient(base_url="https://api.internal.service") as client:
        # X-Request-ID is automatically added to headers
        response = await client.get("/data")
        return response.json()

Sync Usage

from zodiac_core.http import ZodiacSyncClient

def sync_call():
    with ZodiacSyncClient() as client:
        resp = client.get("https://google.com")
        return resp.status_code

Shared Client Resource

When your application uses dependency injection or manages startup/shutdown explicitly, use init_http_client() to create a shared async client resource for the application lifecycle.

from dependency_injector import containers, providers
from zodiac_core.http import init_http_client


class Container(containers.DeclarativeContainer):
    config = providers.Configuration(strict=True)

    external_http_client = providers.Resource(
        init_http_client,
        base_url=config.external.base_url,
        timeout=10.0,
    )

This is useful when a downstream client should receive a preconfigured ZodiacClient, for example with base_url, timeout, or custom event_hooks.

---

3. Manual Context Access

In rare cases where you aren't using ZodiacClient (e.g., using aiohttp or requests), you can manually retrieve the current Request ID.

from zodiac_core.context import get_request_id

request_id = get_request_id()
# Manually pass it to other systems...

For custom ASGI middleware that must set and reset request ID (so it does not leak to the next request), use the request_id_scope(request_id) context manager from zodiac_core.context.

---

4. API Reference

HTTP Clients (Tracing Enabled)

ZodiacClient

Bases: AsyncClient

A wrapper around httpx.AsyncClient that automatically injects the current Trace ID into outgoing requests.

Source code in zodiac_core/http.py

class ZodiacClient(httpx.AsyncClient):
    """
    A wrapper around httpx.AsyncClient that automatically injects
    the current Trace ID into outgoing requests.
    """

    def __init__(
        self,
        *,
        timeout: float = 30.0,
        event_hooks: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ):
        super().__init__(
            timeout=timeout,
            event_hooks=_merge_hooks(event_hooks, _inject_trace_id_async_hook),
            **kwargs,
        )

ZodiacSyncClient

Bases: Client

A wrapper around httpx.Client that automatically injects the current Trace ID into outgoing requests.

Source code in zodiac_core/http.py

class ZodiacSyncClient(httpx.Client):
    """
    A wrapper around httpx.Client that automatically injects
    the current Trace ID into outgoing requests.
    """

    def __init__(
        self,
        *,
        timeout: float = 30.0,
        event_hooks: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ):
        super().__init__(
            timeout=timeout,
            event_hooks=_merge_hooks(event_hooks, _inject_trace_id_hook),
            **kwargs,
        )

init_http_client(*, timeout=30.0, event_hooks=None, **kwargs) async

Create and close a shared ZodiacClient within an application lifecycle.

Source code in zodiac_core/http.py

@asynccontextmanager
async def init_http_client(
    *,
    timeout: float = 30.0,
    event_hooks: Optional[Dict[str, Any]] = None,
    **kwargs: Any,
) -> AsyncGenerator[ZodiacClient, None]:
    """Create and close a shared ZodiacClient within an application lifecycle."""
    async with ZodiacClient(timeout=timeout, event_hooks=event_hooks, **kwargs) as client:
        yield client

Context Utilities

get_request_id()

Retrieve the current Request ID from the context.

Usage

headers = {"X-Request-ID": get_request_id()}
requests.get(url, headers=headers)

Source code in zodiac_core/context.py

def get_request_id() -> Optional[str]:
    """
    Retrieve the current Request ID from the context.

    Usage:
        ```python
        headers = {"X-Request-ID": get_request_id()}
        requests.get(url, headers=headers)
        ```
    """
    return _request_id_ctx_var.get()

request_id_scope(request_id)

Set request_id for the current context and reset on exit (normal or exception). Use in middleware so the id never leaks to the next request.

Source code in zodiac_core/context.py

@contextmanager
def request_id_scope(request_id: str):
    """
    Set request_id for the current context and reset on exit (normal or exception).
    Use in middleware so the id never leaks to the next request.
    """
    token = set_request_id(request_id)
    try:
        yield
    finally:
        reset_request_id(token)