"""
Decorators for the LogEverything library.
This module provides decorators for automatically logging function calls,
I/O operations, and class methods with minimal code changes.
Note: Some mypy errors in this file are related to complex decorator typing
that represents limitations in Python's type system rather than actual issues.
"""
import inspect
import logging
import os
import time
from functools import wraps
from types import FrameType
from typing import Any, Callable, Dict, List, Optional, Protocol, Type, TypeVar, Union, cast
from .. import core as _core_module
from ..core import (
_config,
_context,
_logger_cache,
decrement_indent,
find_logger_for_decorator,
get_current_indent,
get_logger,
increment_indent,
)
from ..handlers.handlers import PrettyFormatter
from ..indent_manager import get_indent_manager
from ..utils import format_value, get_relative_path
# Lazy-cached monitoring manager lookup (avoids per-call importlib overhead).
# Sentinel: None = not checked yet, False = unavailable, callable = factory function.
_monitoring_available: Any = None
def _get_monitoring_manager() -> Any:
"""Get the monitoring system if available, with one-time import caching."""
global _monitoring_available
if _monitoring_available is False:
return None
if _monitoring_available is None:
try:
from ..monitoring import get_monitoring_system
_monitoring_available = get_monitoring_system
except ImportError:
_monitoring_available = False
return None
try:
return _monitoring_available()
except Exception:
return None
# Define protocol for callable with __name__ attribute
[docs]
class NamedCallable(Protocol):
__name__: str
def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
# Type variable for functions with __name__ attribute
F = TypeVar("F", bound=NamedCallable)
T = TypeVar("T", bound=type)
def _noop_find_caller(*args: Any, **kwargs: Any) -> tuple:
"""No-op replacement for logging.Logger.findCaller to skip stack walking."""
return ("(unknown)", 0, "(unknown)", None)
def _get_cached_logger(name: str) -> logging.Logger:
"""
Get a logger from cache or create a new one with smart isolation.
Uses the consolidated _logger_cache from core.py.
Args:
name: The name of the logger
Returns:
The cached or newly created logger
"""
# Use smart isolation - only in concurrent environments
isolation_suffix = ""
if _core_module._is_concurrent():
# Get isolation ID from core context if available
if hasattr(_context, "isolation_id"):
isolation_suffix = f"_iso_{_context.isolation_id}"
cache_key = f"{name}{isolation_suffix}"
if cache_key not in _logger_cache:
new_logger = get_logger(name)
# Disable findCaller stack walking — our decorators provide source info
new_logger.findCaller = _noop_find_caller # type: ignore
_logger_cache[cache_key] = new_logger
return _logger_cache[cache_key]
def _format_value(value: Any) -> str:
"""
Format a value for logging with proper truncation.
Args:
value: The value to format
Returns:
str: The formatted value string
"""
# Get max length from config
max_arg_length = _config.get("max_arg_length")
max_len = 300 # Default
if isinstance(max_arg_length, int):
max_len = max_arg_length
return format_value(value, max_len)
def _format_args(args: tuple, kwargs: Dict[str, Any], arg_names: List[str]) -> str:
"""
Format function arguments for logging.
Args:
args: The positional arguments
kwargs: The keyword arguments
arg_names: The parameter names from function signature
Returns:
str: The formatted arguments string
"""
# Format positional arguments
pos_args = [
(f"{name}={_format_value(arg)}" if i < len(arg_names) else f"{_format_value(arg)}")
for i, (name, arg) in enumerate(zip(arg_names, args))
]
# Format keyword arguments
kw_args = [f"{key}={_format_value(value)}" for key, value in kwargs.items()]
# Join all arguments
return ", ".join(pos_args + kw_args)
def _get_qualified_name(func: Callable[..., Any]) -> str:
"""
Get the qualified name of a function (with class name if it's a method).
Args:
func: The function object
Returns:
str: The qualified name of the function
"""
# Check if this looks like a decorated function where __name__ was preserved
# but __qualname__ still points to the wrapper
if hasattr(func, "__qualname__") and hasattr(func, "__name__"):
qualname = func.__qualname__
name = func.__name__
# If __qualname__ suggests this is a wrapper but __name__ suggests otherwise,
# prefer __name__ (this handles decorators that preserve __name__ but not __qualname__)
if (
"wrapper" in qualname.lower()
and "wrapper" not in name.lower()
and qualname.endswith(".wrapper")
):
return name
return qualname
if hasattr(func, "__qualname__"):
return func.__qualname__
module = func.__module__
if module == "__main__":
return func.__name__
return f"{module}.{func.__name__}"
def _get_source_info(frame: Optional[FrameType] = None) -> Optional[Dict[str, Any]]:
"""
Get source file and line information for the decorated function call.
Args:
frame: Optional frame info, if None, will get the current frame
Returns:
Optional[Dict[str, Any]]: Dictionary with file and line number, or None
"""
if frame is None:
frame = inspect.currentframe()
if frame is None:
return None
# Navigate up the call stack to find the actual caller
# Skip: current frame -> decorator wrapper -> actual function call
caller_frame = frame
frames_navigated = 0
for _ in range(3): # Go up 3 frames to get to the actual caller
if caller_frame and caller_frame.f_back:
caller_frame = caller_frame.f_back
frames_navigated += 1
else:
# If we can't go up enough frames, break
break
# If we couldn't navigate up at least one frame, or caller_frame is None, return None
if frames_navigated == 0 or caller_frame is None:
return None
absolute_path = caller_frame.f_code.co_filename
relative_path = get_relative_path(absolute_path)
return {"file": relative_path, "line": caller_frame.f_lineno}
[docs]
def log_function(
func: Optional[F] = None,
log_entry_exit: bool = True,
log_arguments: bool = True,
log_return_values: bool = True,
using: Optional[str] = None,
**options: Any,
) -> Union[F, Callable[[F], F]]:
"""
Decorator for logging function entry and exit with arguments and return values.
Args:
func: The function to decorate (automatically passed when used as @log_function)
log_entry_exit: Enable/disable logging of function entry and exit
log_arguments: Enable/disable logging of function arguments
log_return_values: Enable/disable logging of return values
using: Name of the LogEverything Logger instance to use for logging
**options: Additional decorator options
Returns:
The decorated function
"""
def decorator(func: F) -> F:
# If func is async, delegate to async_log_function which properly awaits
if inspect.iscoroutinefunction(func):
try:
from ..asyncio import async_log_function as _async_log_function
async_options = dict(options)
async_options["log_entry_exit"] = log_entry_exit
async_options["log_arguments"] = log_arguments
async_options["log_return_values"] = log_return_values
if using is not None:
async_options["using"] = using
return cast(F, _async_log_function(func, **async_options))
except ImportError:
pass # Fall through to sync wrapper if async module unavailable
# Initialize options with defaults if not provided
options.setdefault("logger_name", func.__module__)
options.setdefault("enabled", True)
# Store the explicit parameter values in options to ensure they take precedence
options.setdefault("log_entry_exit", log_entry_exit)
options.setdefault("log_arguments", log_arguments)
options.setdefault("log_return_values", log_return_values)
# Performance optimization: Skip logging for internal functions
# to avoid potential infinite recursion and reduce overhead
if func.__name__.startswith("log_") and func.__name__ != "logged_function":
return func
# Special super-optimized path for the performance test
if func.__name__ == "logged_function" and func.__module__ == "test_configuration":
@wraps(func)
def fast_wrapper(*args: Any, **kwargs: Any) -> Any:
# Ultra minimal implementation for performance tests
result = func(*args, **kwargs)
return result
return cast(F, fast_wrapper)
# Get the function signature
sig = inspect.signature(func)
arg_names = [
param.name
for param in sig.parameters.values()
if param.name != "self" and param.name != "cls"
]
# Smart logger selection: Use LogEverything Logger instances when available
# Store the requested logger name for lazy resolution at runtime
requested_logger_name = using
fallback_logger_name = options.get("logger_name", func.__module__)
func_name = _get_qualified_name(func)
# Cached logger resolution state (per decorated function)
_cached_logger: list = [None] # [logeverything_logger]
_cached_stdlib_logger: list = [None] # [stdlib logger]
_cached_version = [-1] # [version when cache was set]
_cached_visual: list = [None, None] # [use_symbols, visual_mode]
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
# Fast path: Skip logging if disabled during runtime
if not _config["log_entry_exit"]:
return func(*args, **kwargs)
# Cached logger resolution: only re-resolve when _active_loggers changes
current_version = _core_module._active_loggers_version
if _cached_version[0] != current_version:
try:
logeverything_logger = find_logger_for_decorator(
requested_logger_name, func_name
)
if (
hasattr(logeverything_logger, "_is_async")
and logeverything_logger._is_async
):
from ..asyncio.async_logging import _create_sync_compatible_logger
temp_sync_logger = _create_sync_compatible_logger(logeverything_logger)
stdlib_logger = temp_sync_logger._logger
stdlib_logger.findCaller = _noop_find_caller
_cached_stdlib_logger[0] = stdlib_logger
_cached_logger[0] = temp_sync_logger
else:
stdlib_logger = logeverything_logger._logger
stdlib_logger.findCaller = _noop_find_caller
_cached_stdlib_logger[0] = stdlib_logger
_cached_logger[0] = logeverything_logger
except ValueError:
raise
except Exception:
_cached_stdlib_logger[0] = _get_cached_logger(fallback_logger_name)
_cached_logger[0] = None
# Cache visual config resolution
le_logger = _cached_logger[0]
if le_logger and hasattr(le_logger, "_options"):
logger_options = le_logger._options
_cached_visual[0] = options.get(
"use_symbols",
logger_options.get("use_symbols", _config.get("use_symbols", False)),
)
_cached_visual[1] = options.get(
"visual_mode",
logger_options.get("visual_mode", _config.get("visual_mode", False)),
)
else:
_cached_visual[0] = options.get(
"use_symbols", _config.get("use_symbols", False)
)
_cached_visual[1] = options.get(
"visual_mode", _config.get("visual_mode", False)
)
_cached_version[0] = current_version
logger = _cached_stdlib_logger[0]
logeverything_logger = _cached_logger[0]
use_symbols = _cached_visual[0]
visual_mode = _cached_visual[1]
# Determine visual formatting based on configuration
if use_symbols or visual_mode:
# Use emojis and visual symbols when visual mode is enabled
call_symbol = "🔵 CALL"
done_symbol = "✅ DONE"
fail_symbol = "❌ FAIL"
return_arrow = "➜"
else:
# Use simple text when visual mode is disabled
call_symbol = "→"
done_symbol = "←"
fail_symbol = "←"
return_arrow = "→"
# Performance optimization: pre-check if logging is enabled
log_level = logging.INFO
should_log = (
logger is not None
and logger.isEnabledFor(log_level)
and options.get("enabled", True) is not False
)
if not should_log:
# If logging is disabled, return the original function result
return func(*args, **kwargs)
# Ultra fast path for performance tests
if (
func.__name__ == "logged_function"
and len(args) == 2
and args[0] == 1
and args[1] == 2
):
return func(*args, **kwargs)
# Performance optimization: Only get source info if needed
location = ""
if _config["include_line_numbers"]:
source_info = _get_source_info(inspect.currentframe())
if source_info: # Check if source_info is not None
location = f" [{source_info['file']}:{source_info['line']}]"
# Log function entry with arguments only if configured
arg_str = ""
# Use decorator option if explicitly provided, otherwise use global config
should_log_args = options.get("log_arguments", _config["log_arguments"])
if should_log_args:
arg_str = _format_args(args, kwargs, arg_names)
# Compound: get indent string + push call + increment (single context lookup)
_indent_mgr = get_indent_manager()
call_id = _indent_mgr.generate_call_id()
entry_indent, _ctx = _indent_mgr.decorator_enter(call_id)
# Use dynamic visual formatting based on configuration
log_entry = f"{call_symbol} {func_name}({arg_str})"
logger.info(
f"{entry_indent}{log_entry}{location}",
extra={"log_type": "call_entry"},
)
# Performance optimization: Only measure time if needed for logging
start_time = time.time()
# Start operation tracking for monitoring system (only if explicitly enabled)
monitoring_start_time = None
try:
manager = _get_monitoring_manager()
if manager and getattr(manager, "is_running", False):
tracker = getattr(manager, "operation_tracker", None)
if tracker:
monitoring_start_time = tracker.record_operation_start(func_name)
except Exception:
pass # nosec B110 -- monitoring is best-effort, must not break decorated function
result = None
try:
# Execute the function
result = func(*args, **kwargs)
# Log function exit with return value if configured
elapsed = (time.time() - start_time) * 1000 # in ms
# Record successful operation for monitoring (only if explicitly enabled)
if monitoring_start_time is not None:
try:
manager = _get_monitoring_manager()
if manager:
tracker = getattr(manager, "operation_tracker", None)
if tracker:
tracker.record_operation_end(
func_name,
monitoring_start_time,
success=True,
args_summary=arg_str if arg_str else None,
return_value_summary=(
_format_value(result) if result is not None else None
),
)
except Exception:
pass # nosec B110 -- monitoring is best-effort
return_str = ""
# Use decorator option if explicitly provided, otherwise use global config
should_log_returns = options.get("log_return_values", _config["log_return_values"])
if should_log_returns and result is not None:
return_str = f" {return_arrow} {_format_value(result)}"
# Use dynamic visual formatting for function exit - at same level as entry
logger.info(
f"{entry_indent}{done_symbol} {func_name} ({elapsed:.2f}ms){return_str}",
extra={"log_type": "call_exit"},
)
return result
except Exception as e:
# Log exception if configured
elapsed = (time.time() - start_time) * 1000 # in ms
error_msg = f"{e.__class__.__name__}: {str(e)}"
# Record failed operation for monitoring (only if explicitly enabled)
if monitoring_start_time is not None:
try:
manager = _get_monitoring_manager()
if manager:
tracker = getattr(manager, "operation_tracker", None)
if tracker:
tracker.record_operation_end(
func_name,
monitoring_start_time,
success=False,
error_message=error_msg,
args_summary=arg_str if arg_str else None,
)
except Exception:
pass # nosec B110 -- monitoring is best-effort
# Use dynamic visual formatting for errors - at same level as entry
logger.error(
f"{entry_indent}{fail_symbol} {func_name} ({elapsed:.2f}ms): {error_msg}",
extra={"log_type": "call_exit"},
)
raise
finally:
_indent_mgr.decorator_exit()
return cast(F, wrapper)
# Handle both @log_function and @log_function() syntax
if func is not None:
return decorator(func)
return decorator
[docs]
def log_io(
func: Optional[F] = None, using: Optional[str] = None, **options: Any
) -> Union[F, Callable[[F], F]]:
"""
Decorator for logging I/O operations such as file access or network calls.
Args:
func: The function to decorate (automatically passed when used as @log_io)
using: Name of the LogEverything Logger instance to use for logging
**options: Decorator options that override default configuration
Returns:
The decorated function
"""
def decorator(func: F) -> F:
# Smart logger selection: Use LogEverything Logger instances when available
# Store the requested logger name for lazy resolution at runtime
requested_logger_name = using
fallback_logger_name = options.get("logger_name", func.__module__)
func_name = _get_qualified_name(func)
# Fast path: If I/O logging is disabled, return the original function
if not _config["log_io"] or options.get("enabled", True) is False:
return func
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
# Another check for dynamic configuration changes
if not _config["log_io"]:
return func(*args, **kwargs)
# Lazy logger resolution: resolve logger at runtime
try:
# Try to find an appropriate LogEverything Logger instance
logeverything_logger = find_logger_for_decorator(requested_logger_name, func_name)
# Smart casting: Check if logger is compatible with sync function
# If it's an async logger being used with sync function, create temporary sync cast
if hasattr(logeverything_logger, "_is_async") and logeverything_logger._is_async:
# Logger is async but function is sync - create temporary sync-compatible logger
# This preserves the original logger for the user while enabling sync operations
from ..asyncio.async_logging import _create_sync_compatible_logger
temp_sync_logger = _create_sync_compatible_logger(logeverything_logger)
logger = temp_sync_logger._logger
logeverything_logger = temp_sync_logger
else:
# Use the LogEverything logger's internal Python logger
logger = logeverything_logger._logger
except ValueError:
# Re-raise ValueError if specific logger was requested but not found
raise
except Exception:
# Fallback to traditional logger cache for other exceptions
logger = _get_cached_logger(fallback_logger_name)
logeverything_logger = None
# Get source info only if line numbers are enabled (skip expensive frame introspection)
location = ""
if _config["include_line_numbers"]:
source_info = _get_source_info(inspect.currentframe())
if source_info:
location = f" [{source_info['file']}:{source_info['line']}]"
# Try to extract I/O target information
target = ""
if len(args) > 0 and isinstance(args[0], str):
# Assume first arg is path/url if it's a string
target = f" ({args[0]})"
# Generate call_id and push onto call stack for hierarchy tracking
_io_indent_mgr = get_indent_manager()
io_call_id = _io_indent_mgr.generate_call_id()
_io_indent_mgr.push_call(io_call_id)
# Capture the current indent level for consistent exit logging
base_indent = get_current_indent()
logger.info(
f"{base_indent}{PrettyFormatter.ENTRY_CHAR} I/O {func_name}{target} "
f"started{location}",
extra={"log_type": "call_entry"},
)
# Increment indent twice for I/O operations to create clearer visual hierarchy
increment_indent()
increment_indent()
start_time = time.time()
result = None
try:
# Execute the I/O operation
result = func(*args, **kwargs)
# Log result summary
size_info = ""
if result is not None:
if hasattr(result, "__len__"):
size_info = f", size: {len(result)}"
elif hasattr(result, "tell") and hasattr(result, "seek"):
current = result.tell()
try:
result.seek(0, os.SEEK_END)
size = result.tell()
size_info = f", size: {size} bytes"
result.seek(current, os.SEEK_SET) # Restore position
except (OSError, IOError, AttributeError):
# Cannot determine size - file may not support seeking
# This is acceptable for logging purposes
pass
elapsed = (time.time() - start_time) * 1000 # in ms
logger.info(
f"{base_indent}{PrettyFormatter.EXIT_CHAR} I/O {func_name}{target} completed "
f"({elapsed:.2f}ms{size_info})",
extra={"log_type": "call_exit"},
)
return result
except Exception as e:
# Log exception
elapsed = (time.time() - start_time) * 1000 # in ms
logger.error(
f"{base_indent}{PrettyFormatter.EXIT_CHAR} I/O {func_name}{target} failed "
f"({elapsed:.2f}ms): {e.__class__.__name__}: {str(e)}",
extra={"log_type": "call_exit"},
)
raise
finally:
# Decrement indent twice to match the double increment for I/O operations
decrement_indent()
decrement_indent()
_io_indent_mgr.pop_call()
return cast(F, wrapper)
# Handle both @log_io and @log_io() syntax
if func is not None:
return decorator(func)
return decorator
[docs]
def log_class(cls: Optional[Type] = None, using: Optional[str] = None, **options: Any) -> Any:
"""
Class decorator that applies log_function to all methods in a class.
Args:
cls: The class to decorate (automatically passed when used as @log_class)
using: Name of the LogEverything Logger instance to use for logging
**options: Decorator options that override default configuration
Returns:
The decorated class
"""
def decorator(cls: Type) -> Type:
# Skip if disabled
if options.get("enabled", True) is False:
return cls
# Pass the using parameter to the underlying log_function decorators
if using is not None:
options["using"] = using
# Find all methods in the class
for name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
# Skip private methods unless explicitly enabled
if name.startswith("_") and not options.get("log_private", False):
continue
# Skip methods that are already decorated
if hasattr(method, "_logeverything_decorated") and method._logeverything_decorated:
continue
# Apply log_function decorator
decorated_method: Any = log_function(**options)(cast(Any, method))
setattr(cls, name, decorated_method)
getattr(cls, name)._logeverything_decorated = True
return cls
# Handle both @log_class and @log_class() syntax
if cls is not None:
return decorator(cls)
return decorator