| 123456789101112131415161718192021 |
- import contextvars
- import logging
- _tracking_id: contextvars.ContextVar[str] = contextvars.ContextVar("tracking_id", default="")
- def set_tracking_id(task_id: str) -> contextvars.Token:
- return _tracking_id.set(task_id)
- def get_tracking_id() -> str:
- return _tracking_id.get()
- class TrackingFilter(logging.Filter):
- """Logging filter that injects the current tracking ID into every log record."""
- def filter(self, record: logging.LogRecord) -> bool:
- tid = _tracking_id.get()
- record.tracking_id = f"[{tid}] " if tid else ""
- return True
|
