Skip to content

analytics_decorator

Standalone analytics decorator for instance methods.

Extracted from metrics.analytics.Analytics.track_instance_method so that core/ no longer imports the concrete Analytics class at runtime. The decorator relies on duck typing instead of isinstance to remain compatible with any object that exposes execute_async_wrapped_call / execute_sync_wrapped_call.

track_instance_method 

track_instance_method(event_type)

Track instance methods by adding analytics tracking.

Requires the decorated class to expose self.analytics (any object with execute_async_wrapped_call / execute_sync_wrapped_call methods) and an optional self.error_logger.

Parameters:

Name Type Description Default
event_type str

Category name for the tracked event

 required
Source code in src/core/analytics_decorator.py 
 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
106
107
108
109
110
def track_instance_method(event_type: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Track instance methods by adding analytics tracking.

    Requires the decorated class to expose ``self.analytics`` (any object
    with ``execute_async_wrapped_call`` / ``execute_sync_wrapped_call``
    methods) and an optional ``self.error_logger``.

    Args:
        event_type: Category name for the tracked event

    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        """Wrap function with analytics tracking based on sync/async type."""
        is_async = inspect.iscoroutinefunction(func)

        @wraps(func)
        async def async_wrapper(self_arg: Any, *args: Any, **kwargs: Any) -> Any:
            """Wrap async method with analytics tracking."""
            analytics = _get_analytics(self_arg, "execute_async_wrapped_call")
            if analytics is None:
                error_logger = getattr(self_arg, "error_logger", None) or _null_logger()
                error_logger.error(
                    "Analytics missing on %s; %s untracked",
                    self_arg.__class__.__name__,
                    _get_func_name(func),
                )
                return await func(self_arg, *args, **kwargs)

            return await analytics.execute_async_wrapped_call(
                func,
                event_type,
                self_arg,
                *args,
                **kwargs,
            )

        @wraps(func)
        def sync_wrapper(self_arg: Any, *args: Any, **kwargs: Any) -> Any:
            """Wrap sync method with analytics tracking."""
            analytics = _get_analytics(self_arg, "execute_sync_wrapped_call")
            if analytics is None:
                error_logger = getattr(self_arg, "error_logger", None) or _null_logger()
                error_logger.error(
                    "Analytics missing on %s; %s untracked",
                    self_arg.__class__.__name__,
                    _get_func_name(func),
                )
                return func(self_arg, *args, **kwargs)

            return analytics.execute_sync_wrapped_call(
                func,
                event_type,
                self_arg,
                *args,
                **kwargs,
            )

        return async_wrapper if is_async else sync_wrapper

    return decorator