Spaces:
Running
Running
| """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) | |