Spaces:
Running
Running
File size: 3,688 Bytes
d3d0e0e | 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 | """
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)
|