File size: 1,620 Bytes
ba7c9c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | """
Propagates the active LangFuse trace ID into standard Python logging so
application logs can be filtered/correlated by trace_id, independent of the
LangFuse UI.
Uses a contextvar rather than a global so it's safe across concurrent runs on
the same process, and is automatically inherited by any worker thread started
via `contextvars.copy_context().run(...)` (the pattern AnalyzerAgent's
ThreadPoolExecutor already uses for LangFuse's own OTEL context).
"""
import contextvars
import logging
from typing import Optional
_trace_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
"current_trace_id", default=None
)
def set_current_trace_id(trace_id: Optional[str]) -> contextvars.Token:
"""Bind trace_id to the current context. Returns a token for reset_current_trace_id()."""
return _trace_id_var.set(trace_id)
def reset_current_trace_id(token: contextvars.Token) -> None:
"""Restore the contextvar to its prior value, undoing set_current_trace_id()."""
_trace_id_var.reset(token)
def get_current_trace_id() -> Optional[str]:
return _trace_id_var.get()
class TraceIdLogFilter(logging.Filter):
"""
Injects the active trace_id (or "-" if none) into every LogRecord as
record.trace_id.
Must be attached to a logging.Handler, not a Logger: filters on a Logger
only apply to records logged directly through that logger instance, not
to records from child loggers that just propagate up to its handlers.
"""
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = _trace_id_var.get() or "-"
return True
|