"""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)