""" Logging utilities that provide module-specific loggers. """ import logging import sys import inspect from typing import Any, Optional, Callable, TypeVar, cast # Type variable for the decorated function F = TypeVar('F', bound=Callable[..., Any]) def get_logger(name: str = None) -> logging.Logger: """ Get a logger with the given name. If no name is provided, it will use the caller's module name. Args: name: Optional name for the logger. If None, uses the caller's module name. """ if name is None: # Get the caller's module name frame = inspect.currentframe() try: # Get the current module's name parts current_module_parts = __name__.split('.') # Go up the stack to find the first frame not in this module while frame: frame = frame.f_back if not frame: break module = inspect.getmodule(frame) if module: # Get the module's name parts module_parts = module.__name__.split('.') # Check if the module name doesn't end with the current module's name if module_parts[-len(current_module_parts):] != current_module_parts: name = module.__name__ break else: # If we can't find a different module, use the current module's name name = __name__ finally: del frame # Avoid reference cycles logger = logging.getLogger(name) # Only configure the logger if it hasn't been configured yet if not logger.handlers: logger.setLevel(logging.INFO) handler = logging.StreamHandler(sys.stdout) formatter = logging.Formatter( '%(asctime)s [%(name)s] [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) handler.setFormatter(formatter) logger.addHandler(handler) return logger def module_logger(func: F) -> F: """ Decorator that adds a logger to a function. The logger will be named after the module containing the function. """ module = inspect.getmodule(func) logger = get_logger(module.__name__ if module else __name__) def wrapper(*args, **kwargs): return func(logger, *args, **kwargs) return cast(F, wrapper) # Create a module-level logger for this module _logger = get_logger(__name__) def log_info(message: str, **kwargs: Any) -> None: """ Log an info message with optional additional context. The logger will be named after the calling module. Args: message: The message to log **kwargs: Additional context to include in the log """ logger = get_logger() if kwargs: items = ", ".join(f'{k}={v}' for k, v in kwargs.items()) message = f"{message} | {items}" logger.info(message) def log_warning(message: str, **kwargs: Any) -> None: """ Log a warning message with optional additional context. The logger will be named after the calling module. Args: message: The message to log **kwargs: Additional context to include in the log """ logger = get_logger() if kwargs: items = ", ".join(f'{k}={v}' for k, v in kwargs.items()) message = f"{message} | {items}" logger.warning(message) def log_error(message: str, exc_info: Optional[BaseException] = None, **kwargs: Any) -> None: """ Log an error message with optional exception and additional context. The logger will be named after the calling module. Args: message: The message to log exc_info: Optional exception to include in the log **kwargs: Additional context to include in the log """ logger = get_logger() if kwargs: items = ", ".join(f'{k}={v}' for k, v in kwargs.items()) message = f"{message} | {items}" logger.error(message, exc_info=exc_info)