Improved logger

This commit is contained in:
florian committed 2026-10-16 22:22:47 +02:00
1 parent 1eb88e4db2
commit 1be2f45a0c
4 files changed
+91 -20

No files matched your search

+1 -1
View File
@@ -1,4 +1,4 @@
name: Build and Publish Package name: Build and Publish Package via local runner
on: on:
workflow_dispatch: workflow_dispatch:
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "simple-logger-handler" name = "simple-logger-handler"
version = "0.1.0" version = "0.2.0"
description = "Simple reusable logger setup." description = "Simple reusable logger setup."
authors = [{ name = "Florian Gänsejunge" }] authors = [{ name = "Florian Gänsejunge" }]
dependencies = ["python-json-logger"] dependencies = ["python-json-logger"]
+7 -2
View File
@@ -1,3 +1,8 @@
from .logger_handler import setup_logger, LOG_LEVEL from .logger_handler import (
LOG_LEVEL,
adopt_uvicorn_loggers,
configure_logging,
setup_logger,
)
__all__ = ["setup_logger"] __all__ = ["setup_logger", "configure_logging", "adopt_uvicorn_loggers", "LOG_LEVEL"]
+82 -16
View File
@@ -1,20 +1,86 @@
import os
import logging import logging
from pythonjsonlogger import jsonlogger 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
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
if LOG_LEVEL not in {"ERROR", "DEBUG", "INFO", "WARNING", "CRITICAL"}:
LOG_LEVEL = "INFO"
def setup_logger(name: str) -> logging.Logger: def setup_logger(name: str) -> logging.Logger:
logger = logging.getLogger(name) """Return a named logger, ensuring logging is configured."""
if not logger.handlers: configure_logging()
handler = logging.StreamHandler() return logging.getLogger(name)
formatter = jsonlogger.JsonFormatter(
"%(asctime)s %(name)s %(levelname)s %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(getattr(logging, LOG_LEVEL))
logger.debug(f"Logger {name} initialized with level {LOG_LEVEL}")
return logger