Correlation ID HTTP header mock in Python
A lightweight HTTP server that echoes or generates correlation IDs to help test distributed systems.
Python code
29 linesimport json
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer
class CorrelationHandler(BaseHTTPRequestHandler):
CORRELATION_HEADER = "X-Correlation-ID"
def do_GET(self):
correlation_id = self.headers.get(self.CORRELATION_HEADER) or str(uuid.uuid4())
response = {
"path": self.path,
"correlation_id": correlation_id,
"header_present": self.CORRELATION_HEADER in self.headers,
}
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header(self.CORRELATION_HEADER, correlation_id)
self.end_headers()
self.wfile.write(json.dumps(response).encode())
def log_message(self, format, *args):
pass
if __name__ == "__main__":
server = HTTPServer(("localhost", 8080), CorrelationHandler)
print("Mock correlation server running on http://localhost:8080")
server.serve_forever()
Output
$ python correlation_mock.py
Mock correlation server running on http://localhost:8080
# In another terminal:
$ curl -H 'X-Correlation-ID: test-123' http://localhost:8080/api/users
{
"path": "/api/users",
"correlation_id": "test-123",
"header_present": true
}
# Without header:
$ curl http://localhost:8080/health
{
"path": "/health",
"correlation_id": "8b1e7c2e-4f9a-4e0a-b5c2-2f9d9d1b3d6a",
"header_present": false
}
How it works
The handler uses self.headers.get() to read the incoming X-Correlation-ID header, falling back to uuid.uuid4() when it's absent. The response body echoes back the correlation ID as JSON, and the header is also set on the response so clients can trace the request. do_GET is overridden to intercept GET requests, and log_message is silenced to avoid noisy output during testing.
Common mistakes
- Forgetting the `or str(uuid.uuid4())` fallback — responses then fail without an incoming header.
- Missing `send_header` for the correlation ID on the way back, breaking client-side tracing.
- Hardcoding the port instead of reading it from env vars, which breaks parallel test runs.
Variations
- Use `ThreadingHTTPServer` instead of `HTTPServer` for concurrent requests.
- Read an env var like `PORT` to make the mock configurable.
Real-world use cases
- Testing that your service propagates correlation IDs across internal API calls when you can't run the full distributed stack locally.
- Simulating a upstream service for integration tests to verify your client attaches the correct header.
- Echoing a trace context in development so you can visually confirm request correlation without a full tracing tool.
Sponsored
More from Microservices patterns
- BFF aggregation pattern: combine multiple service responses in Python easy
- Backward Compatible Schema Evolution in Python medium
- Bulkhead Thread Pool per Service Mock in Python medium
- CQRS with Separate Read and Write Repositories in Python medium
- Cache-Aside Pattern in Python: Per-Service Mock easy
- Consumer Driven Contract Pact Mock in Python medium
Keep learning
Related tutorials and quizzes for this topic.