How to Add a Correlation ID Tracing Header in Python

A mock middleware generates or preserves a correlation ID header and logs structured JSON messages with it for API request tracing.

Easy Python 3.9+ Aug 9, 2026 API design & gRPC 15 views 0 copies

Python code

41 lines
Python 3.9+
import uuid
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Request:
    headers: dict = field(default_factory=dict)

    def get(self, key, default=None):
        return self.headers.get(key, default)

class CorrelationIdMiddleware:
    def __init__(self, header_name: str = "X-Correlation-ID"):
        self.header_name = header_name

    def process(self, request: Request) -> Request:
        corr_id = request.get(self.header_name)
        if corr_id is None:
            corr_id = str(uuid.uuid4())
            request.headers[self.header_name] = corr_id
        return request

    def log_with_correlation(self, request: Request, message: str) -> str:
        corr_id = request.get(self.header_name, "N/A")
        return json.dumps({
            "correlation_id": corr_id,
            "message": message
        })

if __name__ == "__main__":
    middleware = CorrelationIdMiddleware()
    
    request1 = Request()
    request1 = middleware.process(request1)
    
    request2 = Request(headers={"X-Correlation-ID": "existing-id-123"})
    request2 = middleware.process(request2)
    
    print(middleware.log_with_correlation(request1, "Processing request 1"))
    print(middleware.log_with_correlation(request2, "Processing request 2"))

Output

stdout
{"correlation_id": "5f60a707-8e6e-4d9b-9d24-4b6c126e5a34", "message": "Processing request 1"}
{"correlation_id": "existing-id-123", "message": "Processing request 2"}

How it works

The CorrelationIdMiddleware checks for an existing X-Correlation-ID header; if missing, it generates a new UUID via uuid.uuid4(). This ensures each request has a unique trace identifier across distributed services. The log_with_correlation method uses json.dumps to produce structured logs, making them queryable in log aggregators. The dataclass Request uses field(default_factory=dict) to avoid mutable default arguments, a common Python pitfall.

Common mistakes

  • Forgetting to pass headers in the request constructor, so existing IDs are ignored.
  • Using a mutable default argument like `headers={}` in the dataclass, causing shared state.
  • Not handling case-insensitive headers (real HTTP headers are case-insensitive).

Variations

  1. Use `contextvars` to propagate the correlation ID across async calls without passing the request object.
  2. Generate the ID with `secrets.token_hex(16)` for higher entropy instead of UUID.

Real-world use cases

  • Tracing a user request through microservices by attaching the correlation ID to downstream API calls and logs.
  • Correlating client-side errors to server logs in customer support by including the ID in error responses.
  • Debugging distributed transactions by searching all services for a single correlation ID in log aggregators.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from API design & gRPC

Related tutorials and quizzes for this topic.