core/logging/logger.py

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