""" Structured Logging Configuration - ServiceManagerWeb Configuración de logging estructurado con structlog """ import logging import logging.config import sys from typing import Any, Dict import structlog from app.core.config import get_settings settings = get_settings() def add_correlation_id(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any]: """Agregar correlation ID a los logs si está disponible.""" # En un contexto de request real, esto vendría del middleware # Por ahora es un placeholder return event_dict def configure_structlog(): """Configurar structlog para logging estructurado.""" processors = [ # Add the log level and a timestamp to the event_dict structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), add_correlation_id, ] if settings.LOG_FORMAT == "json": processors.append(structlog.processors.JSONRenderer()) else: processors.append(structlog.dev.ConsoleRenderer()) structlog.configure( processors=processors, wrapper_class=structlog.stdlib.BoundLogger, logger_factory=structlog.stdlib.LoggerFactory(), context_class=dict, cache_logger_on_first_use=True, ) def setup_logging(): """Configurar el sistema de logging completo.""" # Configure structlog configure_structlog() # Configure standard library logging logging_config = { "version": 1, "disable_existing_loggers": False, "formatters": { "json": { "()": structlog.stdlib.ProcessorFormatter, "processor": structlog.processors.JSONRenderer(), }, "console": { "()": structlog.stdlib.ProcessorFormatter, "processor": structlog.dev.ConsoleRenderer(colors=True), }, }, "handlers": { "console": { "level": settings.LOG_LEVEL, "class": "logging.StreamHandler", "stream": sys.stdout, "formatter": "json" if settings.LOG_FORMAT == "json" else "console", }, }, "loggers": { "": { # root logger "handlers": ["console"], "level": settings.LOG_LEVEL, "propagate": False, }, "uvicorn": { "handlers": ["console"], "level": "INFO", "propagate": False, }, "uvicorn.error": { "handlers": ["console"], "level": "INFO", "propagate": False, }, "uvicorn.access": { "handlers": ["console"], "level": "INFO", "propagate": False, }, "sqlalchemy": { "handlers": ["console"], "level": "WARNING", "propagate": False, }, "celery": { "handlers": ["console"], "level": "INFO", "propagate": False, }, }, } # Add file handler if specified if settings.LOG_FILE: logging_config["handlers"]["file"] = { "level": settings.LOG_LEVEL, "class": "logging.handlers.RotatingFileHandler", "filename": settings.LOG_FILE, "maxBytes": 10 * 1024 * 1024, # 10MB "backupCount": 5, "formatter": "json", } # Add file handler to all loggers for logger_config in logging_config["loggers"].values(): logger_config["handlers"].append("file") logging.config.dictConfig(logging_config) # Convenience function to get logger def get_logger(name: str = None) -> structlog.BoundLogger: """ Get a configured structlog logger. Args: name: Logger name (optional) Returns: Configured structlog logger """ return structlog.get_logger(name)