123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""
|
|
Structured logging configuration for the Delphi Database FastAPI app.
|
|
|
|
This module configures structlog to output JSON logs and integrates
|
|
context variables so request-specific fields (e.g., request_id) are
|
|
included automatically in log records.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import logging.config
|
|
from typing import Any, Dict
|
|
|
|
import structlog
|
|
|
|
|
|
def _add_required_defaults(_: Any, __: str, event_dict: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Ensure all required fields exist on every log entry so downstream
|
|
consumers receive a consistent schema.
|
|
"""
|
|
# Required fields per project requirements
|
|
event_dict.setdefault("request_id", None)
|
|
event_dict.setdefault("http.method", None)
|
|
event_dict.setdefault("http.path", None)
|
|
event_dict.setdefault("status_code", None)
|
|
event_dict.setdefault("user.id", None)
|
|
event_dict.setdefault("duration_ms", None)
|
|
return event_dict
|
|
|
|
|
|
def _build_foreign_pre_chain() -> list:
|
|
"""Common processor chain used for stdlib log bridging."""
|
|
return [
|
|
structlog.contextvars.merge_contextvars,
|
|
structlog.stdlib.add_log_level,
|
|
_add_required_defaults,
|
|
structlog.processors.TimeStamper(fmt="iso", key="timestamp"),
|
|
structlog.processors.dict_tracebacks,
|
|
]
|
|
|
|
|
|
def _ensure_stdlib_processor_logging(log_level: int) -> None:
|
|
"""Route stdlib (and uvicorn) loggers through structlog's ProcessorFormatter.
|
|
|
|
This ensures uvicorn error logs and any stdlib logs are emitted as JSON,
|
|
matching the structlog output used by the application code.
|
|
"""
|
|
formatter = structlog.stdlib.ProcessorFormatter(
|
|
processor=structlog.processors.JSONRenderer(),
|
|
foreign_pre_chain=_build_foreign_pre_chain(),
|
|
)
|
|
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(formatter)
|
|
|
|
# Root logger
|
|
root_logger = logging.getLogger()
|
|
root_logger.handlers = [handler]
|
|
root_logger.setLevel(log_level)
|
|
|
|
# Uvicorn loggers (error/access/parent)
|
|
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
|
|
uv_logger = logging.getLogger(name)
|
|
uv_logger.handlers = [handler]
|
|
uv_logger.propagate = False
|
|
uv_logger.setLevel(log_level)
|
|
|
|
|
|
def build_uvicorn_structlog_formatter() -> logging.Formatter:
|
|
"""
|
|
Factory used by logging.config.dictConfig to create a ProcessorFormatter
|
|
that renders JSON and merges contextvars. This is referenced by
|
|
app/uvicorn_log_config.json.
|
|
"""
|
|
return structlog.stdlib.ProcessorFormatter(
|
|
processor=structlog.processors.JSONRenderer(),
|
|
foreign_pre_chain=_build_foreign_pre_chain(),
|
|
)
|
|
|
|
|
|
def setup_logging(log_level: int = logging.INFO) -> None:
|
|
"""
|
|
Configure structlog for JSON logging with contextvars support and bridge
|
|
stdlib/uvicorn loggers to JSON output.
|
|
|
|
Args:
|
|
log_level: Minimum log level for application logs.
|
|
"""
|
|
# Do not clobber handlers if already configured by a log-config (e.g., uvicorn --log-config)
|
|
# basicConfig is a no-op if handlers exist.
|
|
logging.basicConfig(level=log_level)
|
|
|
|
structlog.configure(
|
|
processors=[
|
|
structlog.stdlib.filter_by_level,
|
|
structlog.contextvars.merge_contextvars,
|
|
structlog.stdlib.add_log_level,
|
|
_add_required_defaults,
|
|
structlog.processors.TimeStamper(fmt="iso", key="timestamp"),
|
|
structlog.processors.dict_tracebacks,
|
|
# Defer rendering to logging's ProcessorFormatter to avoid double JSON
|
|
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
|
|
],
|
|
context_class=dict,
|
|
logger_factory=structlog.stdlib.LoggerFactory(),
|
|
wrapper_class=structlog.stdlib.BoundLogger,
|
|
cache_logger_on_first_use=True,
|
|
)
|
|
|
|
# If uvicorn/root handlers are not already using ProcessorFormatter, ensure they do.
|
|
def _has_processor_formatter(logger: logging.Logger) -> bool:
|
|
for h in logger.handlers:
|
|
if isinstance(getattr(h, "formatter", None), structlog.stdlib.ProcessorFormatter):
|
|
return True
|
|
return False
|
|
|
|
if not (_has_processor_formatter(logging.getLogger()) or _has_processor_formatter(logging.getLogger("uvicorn"))):
|
|
_ensure_stdlib_processor_logging(log_level)
|
|
|
|
|