Spaces:
Running
Running
| """ | |
| Centralized logging utility for the data agent baseline. | |
| Simple debug logging that includes filename and function name. | |
| """ | |
| from __future__ import annotations | |
| import inspect | |
| import sys | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any | |
| class AgentLogger: | |
| """Simple logger that respects log_debug configuration flag.""" | |
| def __init__(self, log_debug: bool = False) -> None: | |
| self.log_debug = log_debug | |
| def _get_caller_info(self) -> tuple[str, str]: | |
| """Get the filename and function name of the caller.""" | |
| frame = inspect.currentframe() | |
| if frame is None: | |
| return "unknown", "unknown" | |
| # Go up the stack: _get_caller_info -> _format_message -> debug/info/etc -> actual caller | |
| caller_frame = frame.f_back.f_back.f_back | |
| if caller_frame is None: | |
| return "unknown", "unknown" | |
| filename = Path(caller_frame.f_code.co_filename).name | |
| function_name = caller_frame.f_code.co_name | |
| return filename, function_name | |
| def _format_message(self, level: str, message: str, **kwargs: Any) -> str: | |
| """Format a log message with timestamp, filename, function, and optional metadata.""" | |
| timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] | |
| filename, function_name = self._get_caller_info() | |
| parts = [f"[{timestamp}] [{level}] [{filename}:{function_name}] {message}"] | |
| if kwargs: | |
| metadata = ", ".join(f"{k}={v}" for k, v in kwargs.items()) | |
| parts.append(f" ({metadata})") | |
| return "".join(parts) | |
| def debug(self, message: str, **kwargs: Any) -> None: | |
| """Log a debug message (only if log_debug is enabled).""" | |
| if self.log_debug: | |
| print(self._format_message("DEBUG", message, **kwargs), file=sys.stderr) | |
| def info(self, message: str, **kwargs: Any) -> None: | |
| """Log an info message (only if log_debug is enabled).""" | |
| if self.log_debug: | |
| print(self._format_message("INFO", message, **kwargs), file=sys.stderr) | |
| def warning(self, message: str, **kwargs: Any) -> None: | |
| """Log a warning message (always logged).""" | |
| print(self._format_message("WARNING", message, **kwargs), file=sys.stderr) | |
| def error(self, message: str, **kwargs: Any) -> None: | |
| """Log an error message (always logged).""" | |
| print(self._format_message("ERROR", message, **kwargs), file=sys.stderr) | |
| # Global logger instance (will be initialized by the application) | |
| _global_logger: AgentLogger | None = None | |
| def initialize_logger(log_debug: bool = False) -> None: | |
| """Initialize the global logger with configuration.""" | |
| global _global_logger | |
| _global_logger = AgentLogger(log_debug=log_debug) | |
| def get_logger() -> AgentLogger: | |
| """Get the global logger instance.""" | |
| global _global_logger | |
| if _global_logger is None: | |
| # Default to no logging if not initialized | |
| _global_logger = AgentLogger(log_debug=False) | |
| return _global_logger | |
| def debug(message: str, **kwargs: Any) -> None: | |
| """Convenience function for debug logging.""" | |
| get_logger().debug(message, **kwargs) | |
| def info(message: str, **kwargs: Any) -> None: | |
| """Convenience function for info logging.""" | |
| get_logger().info(message, **kwargs) | |
| def warning(message: str, **kwargs: Any) -> None: | |
| """Convenience function for warning logging.""" | |
| get_logger().warning(message, **kwargs) | |
| def error(message: str, **kwargs: Any) -> None: | |
| """Convenience function for error logging.""" | |
| get_logger().error(message, **kwargs) | |