""" 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)