How to Add a Correlation ID to Logging Records in Python
Attach a unique correlation ID to every log record using a custom logging.Filter, making distributed request tracking traceable.
Python code
29 linesimport logging
import uuid
from dataclasses import dataclass, field
@dataclass
class CorrelationIdFilter(logging.Filter):
correlation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
def filter(self, record: logging.LogRecord) -> bool:
record.correlation_id = self.correlation_id
return True
if __name__ == "__main__":
handler = logging.StreamHandler()
handler.addFilter(CorrelationIdFilter())
formatter = logging.Formatter(
"%(asctime)s [%(correlation_id)s] %(levelname)s: %(message)s"
)
handler.setFormatter(formatter)
logger = logging.getLogger("app")
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)
logger.info("User action completed")
logger.error("Database connection failed")
Output
2025-04-06 12:34:56,789 [3f9c1f2e-8d4b-4a6e-b9c1-2d3c4e5f6a7b] INFO: User action completed
2025-04-06 12:34:56,790 [3f9c1f2e-8d4b-4a6e-b9c1-2d3c4e5f6a7b] ERROR: Database connection failed
How it works
The CorrelationIdFilter is a subclass of logging.Filter that generates a UUID once per filter instance. When the filter's filter method is called for each log record, it sets record.correlation_id to that UUID. Because the filter is attached to the handler, every record passing through the handler gets the same correlation ID for a given application run. The formatter then accesses %(correlation_id)s in the log format string. This works because Filter instances are invoked internally before the record is formatted.
Common mistakes
- Using `uuid.uuid4()` inside the `filter` method, which creates a new ID per log record instead of per filter.
- Forgetting to add the filter to the handler (or logger) where the records are processed.
- Not including the `%(correlation_id)s` placeholder in the formatter, causing a formatting error.
Variations
- Pass an explicit correlation ID from an incoming request (e.g., from HTTP headers) instead of generating one.
- Use `logging.config.dictConfig` to register the filter declaratively for all handlers.
Real-world use cases
- Correlating all log messages for a single HTTP request across services in a microservices architecture.
- Tracking a user's action through a multi-step ETL pipeline to debug data anomalies.
- Matching logs from background workers to the job that triggered them for easier root-cause analysis.
Sponsored
More from Errors & debugging
- Catch RecursionError and Fail Gracefully in Python easy
- Catch ValueError and print friendly message in Python easy
- Collect Multiple Validation Errors in Python Before Raising medium
- Handle ValueError and ZeroDivisionError in Python with try except easy
- How to Assert Preconditions with Descriptive Messages in Python easy
- How to Assert an Invariant After a Complex Transformation in Python easy
Keep learning
Related tutorials and quizzes for this topic.