from __future__ import annotations import logging import logging.config import sys import threading from typing import Literal import structlog _logging_configured: bool = False _logging_lock: threading.Lock = threading.Lock() def setup_logging( log_level: str = "INFO", log_format: Literal["json", "console"] = "json", ) -> None: """Configure structured logging with JSON or console output. Integrates structlog with the standard library logging module so that all loggers (including third-party) produce consistent structured output. Thread-safe and idempotent — safe to call multiple times. Args: log_level: Logging level string (DEBUG, INFO, WARNING, ERROR, CRITICAL). log_format: Output format — "json" for production, "console" for development. """ global _logging_configured with _logging_lock: if _logging_configured: return level = getattr(logging, log_level.upper(), logging.INFO) renderer: structlog.types.Processor if log_format == "console": renderer = structlog.dev.ConsoleRenderer() else: renderer = structlog.processors.JSONRenderer() shared_processors: list[structlog.types.Processor] = [ structlog.contextvars.merge_contextvars, structlog.stdlib.add_log_level, structlog.stdlib.add_logger_name, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.UnicodeDecoder(), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ] structlog.configure( processors=[ *shared_processors, structlog.stdlib.filter_by_level, structlog.stdlib.PositionalArgumentsFormatter(), 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, ) formatter = structlog.stdlib.ProcessorFormatter( processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, renderer, ], foreign_pre_chain=[ structlog.contextvars.merge_contextvars, structlog.stdlib.add_log_level, structlog.stdlib.add_logger_name, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.UnicodeDecoder(), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ], ) logging.config.dictConfig( { "version": 1, "disable_existing_loggers": False, "formatters": { "structlog": { "()": structlog.stdlib.ProcessorFormatter, "processors": [ structlog.stdlib.ProcessorFormatter.remove_processors_meta, renderer, ], "foreign_pre_chain": [ structlog.contextvars.merge_contextvars, structlog.stdlib.add_log_level, structlog.stdlib.add_logger_name, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.UnicodeDecoder(), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, ], }, }, "handlers": { "default": { "class": "logging.StreamHandler", "stream": sys.stdout, "formatter": "structlog", }, }, "root": { "level": level, "handlers": ["default"], }, } ) _logging_configured = True def get_logger(name: str = "app") -> structlog.stdlib.BoundLogger: """Return a structured logger bound with the given name. Args: name: Logger name, typically __name__ of the calling module. """ logger: structlog.stdlib.BoundLogger = structlog.get_logger(name) return logger