65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
import logging
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
class StructuredFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
log_entry = {
|
|
"timestamp": self.formatTime(record),
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"message": record.getMessage(),
|
|
}
|
|
|
|
if hasattr(record, "extra_fields"):
|
|
log_entry.update(record.extra_fields)
|
|
|
|
if record.exc_info:
|
|
log_entry["exception"] = self.formatException(record.exc_info)
|
|
|
|
return str(log_entry)
|
|
|
|
|
|
def setup_logging() -> None:
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(logging.INFO)
|
|
|
|
# Remove default handlers
|
|
for handler in root_logger.handlers[:]:
|
|
root_logger.removeHandler(handler)
|
|
|
|
# Add structured handler
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(StructuredFormatter())
|
|
root_logger.addHandler(handler)
|
|
|
|
# Set levels for third-party loggers
|
|
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
|
|
logging.getLogger("httpx").setLevel(logging.WARNING)
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
return logging.getLogger(name)
|
|
|
|
|
|
class LogContext:
|
|
def __init__(self, logger: logging.Logger, **context: Any):
|
|
self.logger = logger
|
|
self.context = context
|
|
|
|
def info(self, message: str, **extra: Any) -> None:
|
|
self._log(logging.INFO, message, **extra)
|
|
|
|
def warning(self, message: str, **extra: Any) -> None:
|
|
self._log(logging.WARNING, message, **extra)
|
|
|
|
def error(self, message: str, **extra: Any) -> None:
|
|
self._log(logging.ERROR, message, **extra)
|
|
|
|
def _log(self, level: int, message: str, **extra: Any) -> None:
|
|
combined_extra = {**self.context, **extra}
|
|
record = self.logger.makeRecord(
|
|
self.logger.name, level, "", 0, message, (), None, extra_fields=combined_extra
|
|
)
|
|
self.logger.handle(record)
|