Spaces:
Sleeping
Sleeping
File size: 1,917 Bytes
efff1ff | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """Structured JSON logging with request correlation."""
from __future__ import annotations
import logging
import sys
import time
import uuid
from contextvars import ContextVar
from typing import Any
_request_id: ContextVar[str] = ContextVar("request_id", default="-")
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
import json
payload: dict[str, Any] = {
"ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
+ f".{int(record.msecs):03d}Z",
"level": record.levelname,
"logger": record.name,
"msg": record.getMessage(),
"request_id": _request_id.get(),
}
for key in ("document_id", "latency_ms", "status", "n_keys", "detail"):
val = getattr(record, key, None)
if val is not None:
payload[key] = val
if record.exc_info:
payload["exc"] = self.formatException(record.exc_info)
return json.dumps(payload, ensure_ascii=False)
def configure_logging(level: str = "INFO", json_output: bool = True) -> None:
root = logging.getLogger()
root.handlers.clear()
handler = logging.StreamHandler(sys.stdout)
if json_output:
handler.setFormatter(JsonFormatter())
else:
handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)-7s %(name)s | %(message)s")
)
root.addHandler(handler)
root.setLevel(level.upper())
# uvicorn duplicates access logs through its own handlers.
for noisy in ("uvicorn.access", "uvicorn.error"):
logging.getLogger(noisy).handlers.clear()
logging.getLogger(noisy).propagate = True
def new_request_id() -> str:
rid = uuid.uuid4().hex[:16]
_request_id.set(rid)
return rid
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)
|