""" Core Logger Factory and Configuration. This module provides the centralized logger factory that should be used across all GEPA Optimizer modules. It ensures consistent logging behavior and formatting throughout the application. Design Principles: - Single source of truth for logger configuration - Lazy initialization (loggers created on first use) - Thread-safe logger access - Configurable log levels per module """ import logging import sys from enum import Enum from typing import Optional, Dict, Any from functools import lru_cache from .formatters import GepaFormatter # Root logger name for GEPA Optimizer GEPA_LOGGER_NAME = "gepa_optimizer" # Default log format DEFAULT_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s" DEFAULT_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" class LogLevel(str, Enum): """Supported log levels with string representation.""" DEBUG = "DEBUG" INFO = "INFO" WARNING = "WARNING" ERROR = "ERROR" CRITICAL = "CRITICAL" @classmethod def from_string(cls, level: str) -> "LogLevel": """Convert string to LogLevel enum.""" try: return cls(level.upper()) except ValueError: return cls.INFO class LoggerConfig: """ Configuration class for GEPA logging. This class holds all logging configuration and can be modified before calling configure_logging() to customize behavior. """ # Default configuration level: LogLevel = LogLevel.INFO format: str = DEFAULT_FORMAT date_format: str = DEFAULT_DATE_FORMAT # Module-specific log levels (for fine-grained control) module_levels: Dict[str, LogLevel] = {} # Output configuration log_to_console: bool = True log_to_file: Optional[str] = None # Formatting options use_colors: bool = True include_emoji: bool = True # For visual clarity in development @classmethod def reset(cls) -> None: """Reset configuration to defaults.""" cls.level = LogLevel.INFO cls.format = DEFAULT_FORMAT cls.date_format = DEFAULT_DATE_FORMAT cls.module_levels = {} cls.log_to_console = True cls.log_to_file = None cls.use_colors = True cls.include_emoji = True # Global flag to track if logging is configured _logging_configured = False def configure_logging( level: Optional[str] = None, log_file: Optional[str] = None, use_colors: bool = True, include_emoji: bool = True, format_string: Optional[str] = None, module_levels: Optional[Dict[str, str]] = None, ) -> None: """ Configure the GEPA logging system. This should be called once at application startup. Subsequent calls will update the configuration. Args: level: Global log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_file: Optional path to log file use_colors: Whether to use colored output in console include_emoji: Whether to include emoji prefixes for visual clarity format_string: Custom format string (optional) module_levels: Dict mapping module names to their specific log levels Example: configure_logging( level="DEBUG", log_file="optimization.log", module_levels={ "gepa_optimizer.core.optimizer": "INFO", "gepa_optimizer.llms": "DEBUG" } ) """ global _logging_configured # Update configuration if level: LoggerConfig.level = LogLevel.from_string(level) if log_file: LoggerConfig.log_to_file = log_file LoggerConfig.use_colors = use_colors LoggerConfig.include_emoji = include_emoji if format_string: LoggerConfig.format = format_string if module_levels: LoggerConfig.module_levels = { k: LogLevel.from_string(v) for k, v in module_levels.items() } # Get or create root GEPA logger root_logger = logging.getLogger(GEPA_LOGGER_NAME) root_logger.setLevel(getattr(logging, LoggerConfig.level.value)) # Remove existing handlers to avoid duplicates root_logger.handlers.clear() # Console handler if LoggerConfig.log_to_console: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(getattr(logging, LoggerConfig.level.value)) # Use custom formatter formatter = GepaFormatter( fmt=LoggerConfig.format, datefmt=LoggerConfig.date_format, use_colors=use_colors, include_emoji=include_emoji, ) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # File handler (if configured) if LoggerConfig.log_to_file: file_handler = logging.FileHandler(LoggerConfig.log_to_file) file_handler.setLevel(getattr(logging, LoggerConfig.level.value)) # File logs don't use colors file_formatter = GepaFormatter( fmt=LoggerConfig.format, datefmt=LoggerConfig.date_format, use_colors=False, include_emoji=False, ) file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) # Apply module-specific levels for module_name, module_level in LoggerConfig.module_levels.items(): module_logger = logging.getLogger(module_name) module_logger.setLevel(getattr(logging, module_level.value)) _logging_configured = True # Log that configuration is complete root_logger.debug( f"Logging configured: level={LoggerConfig.level.value}, " f"file={LoggerConfig.log_to_file}" ) @lru_cache(maxsize=128) def get_logger(name: str) -> logging.Logger: """ Get a logger instance for the given module name. This is the primary factory function for obtaining loggers. All GEPA modules should use this instead of logging.getLogger(). Args: name: Module name (typically __name__) Returns: Configured Logger instance Example: from gepa_optimizer.infrastructure.logging import get_logger logger = get_logger(__name__) logger.info("Starting process") logger.error("Failed to connect", exc_info=True) """ global _logging_configured # Auto-configure with defaults if not yet configured if not _logging_configured: configure_logging() # Ensure name is under GEPA namespace for consistent handling if not name.startswith(GEPA_LOGGER_NAME) and name != GEPA_LOGGER_NAME: # External module - still use our formatting pass logger = logging.getLogger(name) # Apply module-specific level if configured if name in LoggerConfig.module_levels: logger.setLevel(getattr(logging, LoggerConfig.module_levels[name].value)) return logger def set_log_level(level: str, module: Optional[str] = None) -> None: """ Dynamically change log level at runtime. Args: level: New log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) module: Optional module name. If None, changes global level. Example: # Enable debug for specific module set_log_level("DEBUG", "gepa_optimizer.core.optimizer") # Change global level set_log_level("WARNING") """ log_level = LogLevel.from_string(level) if module: # Set level for specific module logger = logging.getLogger(module) logger.setLevel(getattr(logging, log_level.value)) LoggerConfig.module_levels[module] = log_level else: # Set global level LoggerConfig.level = log_level root_logger = logging.getLogger(GEPA_LOGGER_NAME) root_logger.setLevel(getattr(logging, log_level.value)) # Update all handlers for handler in root_logger.handlers: handler.setLevel(getattr(logging, log_level.value))