FinDataPilot / app /utils /trace.py
Fin-DataPilot Deploy Bot
ci: 5d4fb19
f992c25
Raw
History Blame Contribute Delete
1.51 kB
"""Trace ID generation and structured logging."""
from __future__ import annotations
import logging
import secrets
import sys
from typing import Any
import orjson
def generate_trace_id() -> str:
"""Generate a 64-char hex trace ID."""
return secrets.token_hex(32)
class JsonFormatter(logging.Formatter):
"""Compact JSON log line; pairs trace_id if present in record."""
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
"level": record.levelname,
"logger": record.name,
"msg": record.getMessage(),
}
trace_id = getattr(record, "trace_id", None)
if trace_id:
payload["trace_id"] = trace_id
if record.exc_info:
payload["exc"] = self.formatException(record.exc_info)
return orjson.dumps(payload).decode()
def setup_logging(level: str = "INFO") -> None:
"""Install the JSON formatter on the root logger."""
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.handlers.clear()
root.addHandler(handler)
root.setLevel(getattr(logging, level.upper(), logging.INFO))
# Tame chatty libraries
for noisy in ("httpx", "httpcore", "openai", "urllib3"):
logging.getLogger(noisy).setLevel(logging.WARNING)
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)