File size: 4,086 Bytes
4aec76b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
"""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")
|