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.
Python code
41 linesimport 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
{"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
- Use `contextvars` to propagate the correlation ID across async calls without passing the request object.
- 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
More from API design & gRPC
- Build a Bulk Array POST Mock Server in Python medium
- Build a Mock REST API with PUT and GET in Python medium
- Convert Protobuf to JSON and Dict in Python easy
- Create a Data Helper in Python for gRPC-style APIs easy
- Format data in Python using dataclasses like gRPC messages easy
- Generate an OpenAPI Spec from Mock Routes in Python easy
Keep learning
Related tutorials and quizzes for this topic.