server/logger.py

"""Logger module for centralized logging across the RAG system.

Provides a unified logging interface for all components with:
- File logging with configurable output paths
- Console logging with level filtering
- Timezone support (CST - Central Standard Time)
- Instance caching to prevent duplicate loggers
"""

import logging
from typing import Literal, Optional

# Logs priority levels (highest to lowest):
# CRITICAL > ERROR > WARNING > INFO > DEBUG > NOTSET

# Global cache for logger instances to avoid creating duplicates:
_logger_instances = {}


def get_logger(
    name: str = "unset",
    log_to_console: bool = False,
    log_to_file: bool = True,
    log_file: Optional[str] = "app.log",
) -> logging.Logger:
    """Get or create a logger instance with the given name and configuration.
    
    Uses instance caching to ensure only one logger per name exists, preventing
    duplicate logs from being written.

    Args:
        name (str): Name identifier for the logger (shown in log messages).
        log_to_console (bool): Whether to output logs to console. Defaults to False.
        log_to_file (bool): Whether to output logs to file. Defaults to True.
        log_file (Optional[str]): Path to the log file. If None, file logging is disabled.

    Returns:
        logging.Logger: Configured logger instance for use in code.
        
    Example:
        >>> logger = get_logger("my_module", log_to_console=True)
        >>> logger.info("Application started")
    """

    # Return cached logger if it already exists
    if name in _logger_instances:
        return _logger_instances[name]

    # Create new logger instance
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)  # Capture all log levels
    logger.propagate = False  # Prevent duplicate logs from root logger

    # Format logs with timestamp (CST), level, logger name, and message
    formatter = logging.Formatter(
        '%(asctime)s.%(msecs)03d [%(levelname)-8s] [%(name)s] %(message)s',
        datefmt='%d-%m-%y %H:%M:%S'  # CST timezone format
    )

    # File handler: writes logs to specified file
    if log_to_file and log_file:
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.DEBUG)  # Log all levels to file
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    # Console handler: writes logs to stdout
    if log_to_console:
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.WARNING)  # Only warnings and above to console
        console_handler.setFormatter(formatter)
        logger.addHandler(console_handler)

    # Log initialization details
    logger.info(
        f"Logger initialized. [File: '{log_file if log_to_file else 'Disabled'}', "
        f"Console: '{'Enabled' if log_to_console else 'Disabled'}']"
    )

    # Cache the logger for future requests
    _logger_instances[name] = logger
    return logger


def log_message(
    logger: logging.Logger,
    message: str,
    level: Literal["debug", "info", "warning", "error", "critical"] = "info"
) -> None:
    """Utility function to log messages at various severity levels.
    
    This is a convenience wrapper around logger.debug(), .info(), etc.

    Args:
        logger (logging.Logger): The logger instance to use.
        message (str): The message text to log.
        level (Literal): The logging level to use. Options: debug, info, warning, error, critical.
                        Defaults to 'info'.
                        
    Example:
        >>> log_message(logger, "Database connection failed", "error")
    """

    # Dynamically call the appropriate logging method
    getattr(logger, level)(message)


# Example usage:
if __name__ == "__main__":
    logger = get_logger(name="Test", log_to_console=True, log_to_file=True)
    logger.debug("This is a debug message")
    logger.info("This is an info message")
    logger.warning("This is a warning message")
    logger.error("This is an error message")
    logger.critical("This is a critical message")