Class StructuredLogger

utils/structured_logger.py:26–160

A structured logger that outputs JSON format logs with consistent fields.

class StructuredLogger:
    """A structured logger that outputs JSON format logs with consistent fields."""
    
    def __init__(self, name: str, log_level: LogLevel = LogLevel.INFO, log_file: str = None):
        self.name = name
        self.log_level = log_level
        self.logger = logging.getLogger(name)
        self.logger.setLevel(self._get_logging_level(log_level))
        
        # Create formatter
        formatter = logging.Formatter('%(message)s')
        
        # Create handler
        if log_file:
            # Ensure log directory exists
            log_path = Path(log_file)
            log_path.parent.mkdir(parents=True, exist_ok=True)
            handler = logging.FileHandler(log_file)
        else:
            handler = logging.StreamHandler(sys.stdout)
        
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        
        # For correlation IDs
        self.correlation_id = None
    
    def _get_logging_level(self, log_level: LogLevel) -> int:
        """Convert LogLevel enum to logging module level."""
        level_map = {
            LogLevel.DEBUG: logging.DEBUG,
            LogLevel.INFO: logging.INFO,
            LogLevel.WARNING: logging.WARNING,
            LogLevel.ERROR: logging.ERROR,
            LogLevel.CRITICAL: logging.CRITICAL
        }
        return level_map.get(log_level, logging.INFO)
    
    def _should_log(self, level: LogLevel) -> bool:
        """Check if a log level should be logged based on configured level."""
        return level >= self.log_level
    
    def _format_log(self, log_type: LogType, level: LogLevel, message: str, 
                    correlation_id: str = None, **kwargs) -> str:
        """Format log entry as JSON string."""
        log_entry = {
            "timestamp": datetime.datetime.now(datetime.UTC),
            "log_type": log_type.value,
            "level": level.value,
            "logger": self.name,
            "message": message,
            "correlation_id": correlation_id or self.correlation_id,
            **kwargs
        }
        return json.dumps(log_entry, default=str)
    
    def _log(self, log_type: LogType, level: LogLevel, message: str, 
             correlation_id: str = None, **kwargs):