85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
import logging
|
|
import os
|
|
import threading
|
|
|
|
from pythonjsonlogger.json import JsonFormatter
|
|
|
|
_VALID_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
|
_FORMAT = "%(asctime)s %(name)s %(levelname)s %(message)s"
|
|
_NOISY_LOGGERS = ("urllib3", "httpx", "httpcore", "asyncio", "multipart")
|
|
_UVICORN_LOGGERS = ("uvicorn", "uvicorn.error", "uvicorn.access")
|
|
|
|
_lock = threading.Lock()
|
|
_configured = False
|
|
|
|
|
|
def _resolve_level() -> tuple[str, bool]:
|
|
"""Return (level_name, was_valid). Falls back to INFO if invalid."""
|
|
raw = os.getenv("LOG_LEVEL", "INFO").upper()
|
|
return (raw, True) if raw in _VALID_LEVELS else ("INFO", False)
|
|
|
|
|
|
# Kept for backwards compatibility with `from simple_logger_handler import LOG_LEVEL`
|
|
LOG_LEVEL, _ = _resolve_level()
|
|
|
|
|
|
def _build_formatter() -> logging.Formatter:
|
|
if os.getenv("LOG_FORMAT", "json").lower() == "text":
|
|
return logging.Formatter(_FORMAT) # readable output for local development
|
|
return JsonFormatter(_FORMAT)
|
|
|
|
|
|
def configure_logging(*, force: bool = False) -> None:
|
|
"""Configure the root logger once. Safe to call repeatedly and from threads.
|
|
|
|
Environment variables:
|
|
LOG_LEVEL DEBUG | INFO | WARNING | ERROR | CRITICAL (default INFO)
|
|
LOG_FORMAT json | text (default json)
|
|
"""
|
|
global _configured
|
|
with _lock:
|
|
if _configured and not force:
|
|
return
|
|
|
|
level, valid = _resolve_level()
|
|
root = logging.getLogger()
|
|
|
|
# Remove only handlers we added earlier, so we never clobber
|
|
# handlers installed by pytest, Sentry, etc.
|
|
for h in list(root.handlers):
|
|
if getattr(h, "_shared_handler", False):
|
|
root.removeHandler(h)
|
|
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(_build_formatter())
|
|
handler._shared_handler = True # marker for idempotency
|
|
root.addHandler(handler)
|
|
root.setLevel(level)
|
|
|
|
for name in _NOISY_LOGGERS:
|
|
logging.getLogger(name).setLevel(max(logging.WARNING, root.level))
|
|
|
|
_configured = True
|
|
|
|
if not valid:
|
|
root.warning("Invalid LOG_LEVEL %r, falling back to INFO",
|
|
os.getenv("LOG_LEVEL"))
|
|
|
|
|
|
def adopt_uvicorn_loggers() -> None:
|
|
"""Route uvicorn's logs through the root handler (same format as everything else).
|
|
|
|
Uvicorn installs its own handlers at startup, so call this from your app's
|
|
lifespan, which runs after uvicorn has configured itself.
|
|
"""
|
|
for name in _UVICORN_LOGGERS:
|
|
lg = logging.getLogger(name)
|
|
lg.handlers.clear()
|
|
lg.setLevel(logging.NOTSET) # inherit from root
|
|
lg.propagate = True
|
|
|
|
|
|
def setup_logger(name: str) -> logging.Logger:
|
|
"""Return a named logger, ensuring logging is configured."""
|
|
configure_logging()
|
|
return logging.getLogger(name) |