Decorators

The helpers/decorators subpackage provides function and class decorators for cross-cutting concerns including caching, retry logic, deprecation, timing, tracing, and transaction management.

Decorators are named without a _decorator suffix (retry, timeout, ttl_cache, ...) and detect async def functions automatically. The old names (retry_decorator, async_trace_span, ...) keep working and emit a DeprecationWarning and will be removed in 6.0.0; see the Decorators tutorial.

Cache

Decorator for caching function return values to avoid redundant computation or I/O.

Caching decorator helpers.

archipy.helpers.decorators.ttl_caching.ttl_cache_decorator module-attribute

ttl_cache_decorator = deprecated_alias(
    "ttl_cache_decorator", ttl_cache
)

archipy.helpers.decorators.ttl_caching.CachedFunction

Wrapper class for a cached function with a clear_cache method.

This class wraps a function to provide TTL-based caching. The cache is shared across all instances when used as an instance method decorator.

Example

@ttl_cache(ttl_seconds=60, maxsize=100)
def expensive_function(x: int) -> int:
    return x * 2


# First call executes the function
result = expensive_function(5)  # Returns 10

# Second call returns cached result
result = expensive_function(5)  # Returns 10 (from cache)

# Clear cache manually
expensive_function.clear_cache()

Source code in archipy/helpers/decorators/ttl_caching.py

class CachedFunction[**P, R]:
    """Wrapper class for a cached function with a clear_cache method.

    This class wraps a function to provide TTL-based caching. The cache is shared
    across all instances when used as an instance method decorator.

    Example:
        ```python
        @ttl_cache(ttl_seconds=60, maxsize=100)
        def expensive_function(x: int) -> int:
            return x * 2


        # First call executes the function
        result = expensive_function(5)  # Returns 10

        # Second call returns cached result
        result = expensive_function(5)  # Returns 10 (from cache)

        # Clear cache manually
        expensive_function.clear_cache()
        ```
    """

    def __init__(
        self,
        func: Callable[..., R],
        cache: MutableMapping[Hashable, object],
        instance: object | None = None,
        lock: threading.RLock | None = None,
    ) -> None:
        """Initialize the cached function wrapper.

        Args:
            func: The function to wrap.
            cache: The cache instance to use.
            instance: The instance this method is bound to (for bound methods).
            lock: Lock guarding the cache; shared between bound copies of the same decorated function.
        """
        self._func: Callable[..., R] = func
        self._cache: MutableMapping[Hashable, object] = cache
        self._instance: object | None = instance
        # Calls on the same cache from several threads must not race on the (non thread-safe) TTLCache
        self._lock = lock if lock is not None else threading.RLock()
        # Preserve function metadata
        wraps(func)(self)
        if inspect.iscoroutinefunction(func):
            inspect.markcoroutinefunction(self)

    def __get__(self, obj: object, objtype: type | None = None) -> CachedFunction[P, R]:
        """Support instance methods by implementing descriptor protocol.

        This method caches the bound method in the instance's __dict__ to ensure
        identity consistency (obj.method is obj.method returns True).
        """
        if obj is None:
            return self

        # Cache the bound method in the instance's __dict__ for identity consistency
        func_name = getattr(self._func, "__name__", "cached_method")
        # id(self) keeps same-named cached methods of different classes in one MRO apart
        bound_method_name = f"_cached_{func_name}_{id(self)}"
        if not hasattr(obj, bound_method_name):
            # Create a bound CachedFunction that shares the same cache
            bound_cached: CachedFunction[P, R] = CachedFunction(self._func, self._cache, instance=obj, lock=self._lock)
            # Store in instance __dict__ to maintain identity
            try:
                object.__setattr__(obj, bound_method_name, bound_cached)
            except AttributeError, TypeError:
                # If we can't set the attribute (frozen dataclass, etc.), return a new instance
                return CachedFunction(self._func, self._cache, instance=obj, lock=self._lock)

        return getattr(obj, bound_method_name)

    def _make_key(self, args: tuple[object, ...], kwargs: Mapping[str, object]) -> Hashable | None:
        """Build a hashable key from the call arguments, or None if they cannot be hashed."""
        from cachetools.keys import typedkey

        key = typedkey(*args, **kwargs)
        try:
            hash(key)
        except TypeError:
            return None
        return key

    def __call__(self, *args: object, **kwargs: object) -> R:
        """Call the cached function.

        Calls whose arguments are not hashable bypass the cache instead of failing.

        Args:
            *args: Positional arguments to pass to the function.
            **kwargs: Keyword arguments to pass to the function.

        Returns:
            The result of the function call (from cache or fresh). For ``async def`` functions
            the result is an awaitable and the awaited value is what gets cached.
        """
        key = self._make_key(args, kwargs)
        call_args = args if self._instance is None else (self._instance, *args)

        if inspect.iscoroutinefunction(self._func):
            # R is the coroutine type here; the cached coroutine wrapper stands in for it
            return cast("R", self._call_async(key, call_args, kwargs))

        if key is None:
            return self._func(*call_args, **kwargs)
        with self._lock:
            cached = self._cache.get(key, _MISSING)
        if cached is not _MISSING:
            return cast("R", cached)

        result: R = self._func(*call_args, **kwargs)
        with self._lock:
            self._cache[key] = result
        return result

    async def _call_async(
        self,
        key: Hashable | None,
        call_args: tuple[object, ...],
        kwargs: Mapping[str, object],
    ) -> object:
        """Await the wrapped coroutine function and cache its awaited value."""
        if key is None:
            return await cast("Awaitable[object]", self._func(*call_args, **kwargs))
        with self._lock:
            cached = self._cache.get(key, _MISSING)
        if cached is not _MISSING:
            return cached

        result = await cast("Awaitable[object]", self._func(*call_args, **kwargs))
        with self._lock:
            self._cache[key] = result
        return result

    def clear_cache(self) -> None:
        """Clear the cache.

        This clears all cached values for this function. When used with instance methods,
        this clears the shared cache for all instances.
        """
        with self._lock:
            self._cache.clear()

archipy.helpers.decorators.ttl_caching.CachedFunction.clear_cache

clear_cache() -> None

Clear the cache.

This clears all cached values for this function. When used with instance methods, this clears the shared cache for all instances.

Source code in archipy/helpers/decorators/ttl_caching.py

def clear_cache(self) -> None:
    """Clear the cache.

    This clears all cached values for this function. When used with instance methods,
    this clears the shared cache for all instances.
    """
    with self._lock:
        self._cache.clear()

archipy.helpers.decorators.ttl_caching.ttl_cache

ttl_cache(
    ttl_seconds: int | Callable[[], int] = 300,
    maxsize: int = 100,
) -> Callable[[Callable[P, R]], CachedFunction[P, R]]

Decorator that provides a TTL cache for functions and methods.

The cache is shared across all instances when decorating instance methods. This is by design to allow efficient caching of expensive operations that depend only on the method arguments, not the instance state.

Parameters:

Name	Type	Description	Default
ttl_seconds	int | Callable[[], int]	Time to live in seconds (default: 5 minutes), or a zero-argument callable that returns the TTL (evaluated once when the decorator is applied). After this time, cached entries expire and the function is re-executed.	300
maxsize	int	Maximum size of the cache (default: 100). When the cache is full, the least recently used entry is evicted.	100

Returns:

Type	Description
Callable[[Callable[P, R]], CachedFunction[P, R]]	Decorated function with TTL caching and a clear_cache() method.

Raises:

Type	Description
ConfigurationError	If cachetools is not installed (archipy[cache] extra).

Example

class DataService:
    @ttl_cache(ttl_seconds=60, maxsize=50)
    def fetch_data(self, key: str) -> dict:
        # Expensive operation
        return {"data": key}


service1 = DataService()
service2 = DataService()

# First call executes the function
result1 = service1.fetch_data("key1")

# Second call from different instance returns cached result
result2 = service2.fetch_data("key1")  # From cache

# Clear cache manually
service1.fetch_data.clear_cache()

Note

• Exceptions are not cached; the function will be re-executed on the next call
• None values are cached like any other value
• Argument types are part of the key (f(1) and f(1.0) are cached separately) and arguments must be hashable; calls with unhashable arguments run uncached
• async def functions are supported; the awaited result is cached
• Access to the cache is guarded by a lock, so it is safe to call from several threads
• Cache is shared across all instances of a class (not per-instance)

Source code in archipy/helpers/decorators/ttl_caching.py

def ttl_cache[**P, R](
    ttl_seconds: int | Callable[[], int] = 300,
    maxsize: int = 100,
) -> Callable[[Callable[P, R]], CachedFunction[P, R]]:
    """Decorator that provides a TTL cache for functions and methods.

    The cache is shared across all instances when decorating instance methods.
    This is by design to allow efficient caching of expensive operations that
    depend only on the method arguments, not the instance state.

    Args:
        ttl_seconds: Time to live in seconds (default: 5 minutes), or a zero-argument
            callable that returns the TTL (evaluated once when the decorator is applied).
            After this time, cached entries expire and the function is re-executed.
        maxsize: Maximum size of the cache (default: 100).
            When the cache is full, the least recently used entry is evicted.

    Returns:
        Decorated function with TTL caching and a clear_cache() method.

    Raises:
        ConfigurationError: If ``cachetools`` is not installed (``archipy[cache]`` extra).

    Example:
        ```python
        class DataService:
            @ttl_cache(ttl_seconds=60, maxsize=50)
            def fetch_data(self, key: str) -> dict:
                # Expensive operation
                return {"data": key}


        service1 = DataService()
        service2 = DataService()

        # First call executes the function
        result1 = service1.fetch_data("key1")

        # Second call from different instance returns cached result
        result2 = service2.fetch_data("key1")  # From cache

        # Clear cache manually
        service1.fetch_data.clear_cache()
        ```

    Note:
        - Exceptions are not cached; the function will be re-executed on the next call
        - None values are cached like any other value
        - Argument types are part of the key (``f(1)`` and ``f(1.0)`` are cached separately) and
          arguments must be hashable; calls with unhashable arguments run uncached
        - ``async def`` functions are supported; the awaited result is cached
        - Access to the cache is guarded by a lock, so it is safe to call from several threads
        - Cache is shared across all instances of a class (not per-instance)
    """
    _require_cachetools()
    from cachetools import TTLCache

    resolved_ttl = int(ttl_seconds) if isinstance(ttl_seconds, int) else int(ttl_seconds())
    cache: TTLCache[Hashable, object, float] = TTLCache(maxsize=maxsize, ttl=resolved_ttl)
    lock = threading.RLock()

    def decorator(func: Callable[P, R]) -> CachedFunction[P, R]:
        return CachedFunction(func, cache, instance=None, lock=lock)

    return decorator

options: show_root_toc_entry: false heading_level: 3

Retry

Decorator that automatically retries a failing function call with configurable backoff strategies.

Retry decorator helpers.

archipy.helpers.decorators.retry_policy.logger module-attribute

logger = logging.getLogger(
    "archipy.helpers.decorators.retry"
)

archipy.helpers.decorators.retry_policy.retry_decorator module-attribute

retry_decorator = deprecated_alias("retry_decorator", retry)

archipy.helpers.decorators.retry_policy.retry

retry(
    max_retries: int = 3,
    delay: float = 1,
    retry_on: tuple[type[Exception], ...] | None = None,
    ignore: tuple[type[Exception], ...] | None = None,
    resource_type: str | None = None,
    backoff: float = 1.0,
    jitter: float = 0.0,
) -> Callable[[Callable[P, R]], Callable[P, R]]

A decorator that retries a function when it raises an exception.

Works for both regular and async def functions. Async functions are awaited and wait with asyncio.sleep between attempts, so the event loop is never blocked.

Parameters:

Name	Type	Description	Default
max_retries	int	The maximum number of attempts, including the first call (3 means one call plus two retries). Must be at least 1. Defaults to 3.	3
delay	float	The delay (in seconds) before the first retry. Defaults to 1.	1
retry_on	Optional[Tuple[Type[Exception], ...]]	A tuple of errors to retry on. If None, retries on all errors. Defaults to None.	None
ignore	Optional[Tuple[Type[Exception], ...]]	A tuple of errors to ignore (not retry on). If None, no errors are ignored. Takes precedence over retry_on. Defaults to None.	None
resource_type	Optional[str]	The type of resource being exhausted. Defaults to None.	None
backoff	float	Multiplier applied to the delay after every failed attempt. 1.0 keeps a fixed delay, 2.0 doubles it each time. Defaults to 1.0.	1.0
jitter	float	Fraction of the delay added randomly (0.1 adds up to 10%) to avoid synchronized retries from many callers. Defaults to 0.0.	0.0

Returns:

Name	Type	Description
Callable	Callable[[Callable[P, R]], Callable[P, R]]	The decorated function with retry logic.

Raises:

Type	Description
InvalidArgumentError	If max_retries is below 1, delay or jitter is negative, or backoff is below 1.

Example

To use this decorator, apply it to a function:

@retry(max_retries=3, delay=1, backoff=2, retry_on=(ValueError,), resource_type="API")
def unreliable_function():
    if random.random() < 0.5:
        raise ValueError("Temporary failure")
    return "Success"


result = unreliable_function()

Output:

2023-10-10 12:00:00,000 - WARNING - Attempt 1 failed: Temporary failure
2023-10-10 12:00:01,000 - INFO - Attempt 2 succeeded.
Success

Source code in archipy/helpers/decorators/retry_policy.py

def retry[**P, R](
    max_retries: int = 3,
    delay: float = 1,
    retry_on: tuple[type[Exception], ...] | None = None,
    ignore: tuple[type[Exception], ...] | None = None,
    resource_type: str | None = None,
    backoff: float = 1.0,
    jitter: float = 0.0,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator that retries a function when it raises an exception.

    Works for both regular and ``async def`` functions. Async functions are awaited and
    wait with ``asyncio.sleep`` between attempts, so the event loop is never blocked.

    Args:
        max_retries (int): The maximum number of attempts, including the first call
            (``3`` means one call plus two retries). Must be at least 1. Defaults to 3.
        delay (float): The delay (in seconds) before the first retry. Defaults to 1.
        retry_on (Optional[Tuple[Type[Exception], ...]]): A tuple of errors to retry on.
            If None, retries on all errors. Defaults to None.
        ignore (Optional[Tuple[Type[Exception], ...]]): A tuple of errors to ignore (not retry on).
            If None, no errors are ignored. Takes precedence over ``retry_on``. Defaults to None.
        resource_type (Optional[str]): The type of resource being exhausted. Defaults to None.
        backoff (float): Multiplier applied to the delay after every failed attempt.
            ``1.0`` keeps a fixed delay, ``2.0`` doubles it each time. Defaults to 1.0.
        jitter (float): Fraction of the delay added randomly (``0.1`` adds up to 10%) to avoid
            synchronized retries from many callers. Defaults to 0.0.

    Returns:
        Callable: The decorated function with retry logic.

    Raises:
        InvalidArgumentError: If ``max_retries`` is below 1, ``delay`` or ``jitter`` is negative,
            or ``backoff`` is below 1.

    Example:
        To use this decorator, apply it to a function:

        ```python
        @retry(max_retries=3, delay=1, backoff=2, retry_on=(ValueError,), resource_type="API")
        def unreliable_function():
            if random.random() < 0.5:
                raise ValueError("Temporary failure")
            return "Success"


        result = unreliable_function()
        ```

        Output:
        ```
        2023-10-10 12:00:00,000 - WARNING - Attempt 1 failed: Temporary failure
        2023-10-10 12:00:01,000 - INFO - Attempt 2 succeeded.
        Success
        ```
    """
    if max_retries < 1:
        raise InvalidArgumentError(argument_name="max_retries", additional_data={"minimum": 1})
    if delay < 0:
        raise InvalidArgumentError(argument_name="delay", additional_data={"minimum": 0})
    if backoff < 1:
        raise InvalidArgumentError(argument_name="backoff", additional_data={"minimum": 1})
    if jitter < 0:
        raise InvalidArgumentError(argument_name="jitter", additional_data={"minimum": 0})

    policy = _RetryPolicy(max_retries, delay, backoff, jitter, retry_on, ignore, resource_type)

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        if inspect.iscoroutinefunction(func):
            # R is the coroutine type here; the async wrapper awaits it and returns the awaited value
            return cast("Callable[P, R]", _wrap_async(cast("Callable[P, Awaitable[object]]", func), policy))
        return _wrap_sync(func, policy)

    return decorator

options: show_root_toc_entry: false heading_level: 3

Timeout

Decorator that enforces a maximum execution time on a function call.

Timeout decorator helpers.

archipy.helpers.decorators.timeout_limit.timeout_decorator module-attribute

timeout_decorator = deprecated_alias(
    "timeout_decorator", timeout
)

archipy.helpers.decorators.timeout_limit.timeout

timeout(
    seconds: float,
) -> Callable[[Callable[P, R]], Callable[P, R]]

A decorator that adds a timeout to a function.

If the function takes longer than the specified number of seconds to execute, a DeadlineExceededError is raised. The timeout is implemented with signal.setitimer(signal.ITIMER_REAL, ...), so fractional seconds are supported. After the call, the previous SIGALRM handler is restored and any timer that was pending before the call is re-armed with its remaining time.

async def functions are supported too: they are bounded with asyncio.timeout, which works in any thread and needs no signals.

Limitations

Synchronous functions rely on SIGALRM, so they work only on Unix and only when the decorated function is called from the main thread of the main interpreter. Calling one from another thread raises FailedPreconditionError instead of running without a timeout. It cannot interrupt blocking C calls that do not return to the interpreter.

Parameters:

Name	Type	Description	Default
seconds	float	The maximum number of seconds the function is allowed to run.	required

Returns:

Name	Type	Description
Callable	Callable[[Callable[P, R]], Callable[P, R]]	The decorated function with a timeout.

Example

To use this decorator, apply it to any function and specify the timeout in seconds:

@timeout(3)  # Set a timeout of 3 seconds
def long_running_function():
    time.sleep(5)  # This will take longer than the timeout
    return "Finished"


try:
    result = long_running_function()
except DeadlineExceededError as e:
    print(e.additional_data)  # {"timeout": 3, "operation": "long_running_function"}

Source code in archipy/helpers/decorators/timeout_limit.py

def timeout[**P, R](seconds: float) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator that adds a timeout to a function.

    If the function takes longer than the specified number of seconds to execute,
    a `DeadlineExceededError` is raised. The timeout is implemented with
    `signal.setitimer(signal.ITIMER_REAL, ...)`, so fractional seconds are supported.
    After the call, the previous SIGALRM handler is restored and any timer that was
    pending before the call is re-armed with its remaining time.

    ``async def`` functions are supported too: they are bounded with `asyncio.timeout`, which
    works in any thread and needs no signals.

    Limitations:
        Synchronous functions rely on `SIGALRM`, so they work only on Unix and only when the
        decorated function is called from the main thread of the main interpreter. Calling one
        from another thread raises `FailedPreconditionError` instead of running without a timeout.
        It cannot interrupt blocking C calls that do not return to the interpreter.

    Args:
        seconds (float): The maximum number of seconds the function is allowed to run.

    Returns:
        Callable: The decorated function with a timeout.

    Example:
        To use this decorator, apply it to any function and specify the timeout in seconds:

        ```python
        @timeout(3)  # Set a timeout of 3 seconds
        def long_running_function():
            time.sleep(5)  # This will take longer than the timeout
            return "Finished"


        try:
            result = long_running_function()
        except DeadlineExceededError as e:
            print(e.additional_data)  # {"timeout": 3, "operation": "long_running_function"}
        ```
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        # Capture function name before wrapping - use getattr for type safety
        func_name = getattr(func, "__name__", "unknown")

        if inspect.iscoroutinefunction(func):
            # R is the coroutine type here; the wrapper awaits it and returns the awaited value
            async_func = cast("Callable[P, Awaitable[object]]", func)

            @wraps(func)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> object:
                try:
                    async with asyncio.timeout(seconds):
                        return await async_func(*args, **kwargs)
                except TimeoutError as e:
                    raise DeadlineExceededError(timeout=seconds, operation=func_name) from e

            return cast("Callable[P, R]", async_wrapper)

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            if threading.current_thread() is not threading.main_thread():
                raise FailedPreconditionError(
                    precondition="timeout requires the main thread",
                    additional_data={"operation": func_name},
                )

            def handle_timeout(_signum: int, _frame: FrameType | None) -> None:
                raise DeadlineExceededError(timeout=seconds, operation=func_name)

            previous_handler = signal.signal(signal.SIGALRM, handle_timeout)
            started = time.monotonic()
            previous_remaining, previous_interval = signal.setitimer(signal.ITIMER_REAL, seconds)

            try:
                return func(*args, **kwargs)
            finally:
                # Disable our timer, restore the previous handler and any timer that was pending
                signal.setitimer(signal.ITIMER_REAL, 0)
                signal.signal(signal.SIGALRM, previous_handler)
                if previous_remaining > 0:
                    remaining = max(previous_remaining - (time.monotonic() - started), 1e-6)
                    signal.setitimer(signal.ITIMER_REAL, remaining, previous_interval)

        return wrapper

    return decorator

options: show_root_toc_entry: false heading_level: 3

Timing

Decorator that measures and records the execution time of a function.

Timing and performance decorator helpers.

archipy.helpers.decorators.timing_log.logger module-attribute

logger = logging.getLogger(
    "archipy.helpers.decorators.timing"
)

archipy.helpers.decorators.timing_log.timing_decorator module-attribute

timing_decorator = deprecated_alias(
    "timing_decorator", timing
)

archipy.helpers.decorators.timing_log.timing

timing(func: Callable[P, R]) -> Callable[P, R]

A decorator that measures the execution time of a function and logs it when DEBUG logging is enabled.

The duration is also logged when the function raises an exception. Works for both regular and async def functions; for async functions the time spent awaiting is included.

Parameters:

Name	Type	Description	Default
func	Callable	The function to be decorated.	required

Returns:

Name	Type	Description
Callable	Callable[P, R]	The wrapped function which logs the execution time when DEBUG logging is enabled.

Example

To use this decorator, simply apply it to any function. For example:

@timing
def example_function(n: int) -> str:
    time.sleep(n)
    return f"Slept for {n} seconds"


result = example_function(2)

Output (if DEBUG logging is enabled):

2023-10-10 12:00:00,000 - DEBUG - example_function took 2.0001 seconds to execute.
Slept for 2 seconds

Source code in archipy/helpers/decorators/timing_log.py

def timing[**P, R](func: Callable[P, R]) -> Callable[P, R]:
    """A decorator that measures the execution time of a function and logs it when DEBUG logging is enabled.

    The duration is also logged when the function raises an exception. Works for both regular
    and ``async def`` functions; for async functions the time spent awaiting is included.

    Args:
        func (Callable): The function to be decorated.

    Returns:
        Callable: The wrapped function which logs the execution time when DEBUG logging is enabled.

    Example:
        To use this decorator, simply apply it to any function. For example:

        ```python
        @timing
        def example_function(n: int) -> str:
            time.sleep(n)
            return f"Slept for {n} seconds"


        result = example_function(2)
        ```

        Output (if DEBUG logging is enabled):
        ```
        2023-10-10 12:00:00,000 - DEBUG - example_function took 2.0001 seconds to execute.
        Slept for 2 seconds
        ```
    """
    # Capture function name before wrapping - use getattr for type safety
    func_name = getattr(func, "__name__", "unknown")

    if inspect.iscoroutinefunction(func):
        # R is the coroutine type here; the wrapper awaits it and returns the awaited value
        async_func = cast("Callable[P, Awaitable[object]]", func)

        @wraps(func)
        async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> object:
            if not logger.isEnabledFor(logging.DEBUG):
                return await async_func(*args, **kwargs)
            start_time = time.perf_counter()
            try:
                return await async_func(*args, **kwargs)
            finally:
                logger.debug("%s took %.4f seconds to execute.", func_name, time.perf_counter() - start_time)

        return cast("Callable[P, R]", async_wrapper)

    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        if not logger.isEnabledFor(logging.DEBUG):
            return func(*args, **kwargs)
        start_time = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            logger.debug("%s took %.4f seconds to execute.", func_name, time.perf_counter() - start_time)

    return wrapper

options: show_root_toc_entry: false heading_level: 3

Tracing

OpenTelemetry span decorators for functions and classes (trace_span, trace_root, trace_class). The old async_trace_span and async_trace_root names are deprecated aliases that will be removed in 6.0.0.

OpenTelemetry tracing decorators for function and class instrumentation.

archipy.helpers.decorators.tracing.trace_span

trace_span(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _SpanDecorator

Decorate a sync or async function with an OpenTelemetry child span.

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	SpanKind | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	Mapping[str, object] | None	Optional static attributes set on the span.	None
capture_args	list[str] | None	Parameter names whose call values are recorded as attributes. Sensitive names (password, token, …) are redacted to ***.	None

Returns:

Type	Description
_SpanDecorator	A decorator that wraps the target function in a span.

Example

@trace_span(name="load_user", capture_args=["user_id"])
def load_user(user_id: int) -> dict[str, Any]:
    return {"id": user_id}

Source code in archipy/helpers/decorators/tracing.py

def trace_span(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _SpanDecorator:
    """Decorate a sync or async function with an OpenTelemetry child span.

    Args:
        name: Span name. Defaults to the function name.
        kind: OpenTelemetry ``SpanKind``. Defaults to ``SpanKind.INTERNAL``.
        attributes: Optional static attributes set on the span.
        capture_args: Parameter names whose call values are recorded as attributes.
            Sensitive names (password, token, …) are redacted to ``***``.

    Returns:
        A decorator that wraps the target function in a span.

    Example:
        ```python
        @trace_span(name="load_user", capture_args=["user_id"])
        def load_user(user_id: int) -> dict[str, Any]:
            return {"id": user_id}
        ```
    """

    def decorator[**P, R](func: Callable[P, R]) -> Callable[P, R]:
        return _traced_wrapper(
            func,
            name=name,
            kind=kind,
            attributes=attributes,
            capture_args=capture_args,
            root=False,
        )

    return decorator

archipy.helpers.decorators.tracing.async_trace_span

async_trace_span(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _AsyncSpanDecorator

Decorate an async function with an OpenTelemetry child span.

Deprecated

Use :func:trace_span, which supports async functions directly.

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	SpanKind | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	Mapping[str, object] | None	Optional static attributes set on the span.	None
capture_args	list[str] | None	Parameter names whose call values are recorded as attributes. Sensitive names (password, token, …) are redacted to ***.	None

Returns:

Type	Description
_AsyncSpanDecorator	A decorator that wraps the target coroutine function in a span.

Raises:

Type	Description
InvalidArgumentError	If the decorated object is not a coroutine function.

Source code in archipy/helpers/decorators/tracing.py

def async_trace_span(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _AsyncSpanDecorator:
    """Decorate an async function with an OpenTelemetry child span.

    Deprecated:
        Use :func:`trace_span`, which supports async functions directly.

    Args:
        name: Span name. Defaults to the function name.
        kind: OpenTelemetry ``SpanKind``. Defaults to ``SpanKind.INTERNAL``.
        attributes: Optional static attributes set on the span.
        capture_args: Parameter names whose call values are recorded as attributes.
            Sensitive names (password, token, …) are redacted to ``***``.

    Returns:
        A decorator that wraps the target coroutine function in a span.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """
    warn_deprecated_name("async_trace_span", "trace_span")

    def decorator[**P, C: Coroutine[object, Never, object]](func: _NamedFunction[P, C]) -> Callable[P, C]:
        _require_coroutine_function(func, "async_trace_span")
        return _traced_wrapper(
            func,
            name=name,
            kind=kind,
            attributes=attributes,
            capture_args=capture_args,
            root=False,
        )

    return decorator

archipy.helpers.decorators.tracing.trace_root

trace_root(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _SpanDecorator

Decorate a sync or async function with a root OpenTelemetry span.

Starts the span with a fresh context detached from any parent span (INVALID_SPAN context).

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	SpanKind | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	Mapping[str, object] | None	Optional static attributes set on the span.	None
capture_args	list[str] | None	Parameter names whose call values are recorded as attributes. Sensitive names (password, token, …) are redacted to ***.	None

Returns:

Type	Description
_SpanDecorator	A decorator that wraps the target function in a root span.

Source code in archipy/helpers/decorators/tracing.py

def trace_root(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _SpanDecorator:
    """Decorate a sync or async function with a root OpenTelemetry span.

    Starts the span with a fresh context detached from any parent span
    (``INVALID_SPAN`` context).

    Args:
        name: Span name. Defaults to the function name.
        kind: OpenTelemetry ``SpanKind``. Defaults to ``SpanKind.INTERNAL``.
        attributes: Optional static attributes set on the span.
        capture_args: Parameter names whose call values are recorded as attributes.
            Sensitive names (password, token, …) are redacted to ``***``.

    Returns:
        A decorator that wraps the target function in a root span.
    """

    def decorator[**P, R](func: Callable[P, R]) -> Callable[P, R]:
        return _traced_wrapper(
            func,
            name=name,
            kind=kind,
            attributes=attributes,
            capture_args=capture_args,
            root=True,
        )

    return decorator

archipy.helpers.decorators.tracing.async_trace_root

async_trace_root(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _AsyncSpanDecorator

Decorate an async function with a root OpenTelemetry span.

Deprecated

Use :func:trace_root, which supports async functions directly.

Starts the span with a fresh context detached from any parent span (INVALID_SPAN context).

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	SpanKind | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	Mapping[str, object] | None	Optional static attributes set on the span.	None
capture_args	list[str] | None	Parameter names whose call values are recorded as attributes. Sensitive names (password, token, …) are redacted to ***.	None

Returns:

Type	Description
_AsyncSpanDecorator	A decorator that wraps the target coroutine function in a root span.

Raises:

Type	Description
InvalidArgumentError	If the decorated object is not a coroutine function.

Source code in archipy/helpers/decorators/tracing.py

def async_trace_root(
    name: str | None = None,
    *,
    kind: SpanKind | None = None,
    attributes: Mapping[str, object] | None = None,
    capture_args: list[str] | None = None,
) -> _AsyncSpanDecorator:
    """Decorate an async function with a root OpenTelemetry span.

    Deprecated:
        Use :func:`trace_root`, which supports async functions directly.

    Starts the span with a fresh context detached from any parent span
    (``INVALID_SPAN`` context).

    Args:
        name: Span name. Defaults to the function name.
        kind: OpenTelemetry ``SpanKind``. Defaults to ``SpanKind.INTERNAL``.
        attributes: Optional static attributes set on the span.
        capture_args: Parameter names whose call values are recorded as attributes.
            Sensitive names (password, token, …) are redacted to ``***``.

    Returns:
        A decorator that wraps the target coroutine function in a root span.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """
    warn_deprecated_name("async_trace_root", "trace_root")

    def decorator[**P, C: Coroutine[object, Never, object]](func: _NamedFunction[P, C]) -> Callable[P, C]:
        _require_coroutine_function(func, "async_trace_root")
        return _traced_wrapper(
            func,
            name=name,
            kind=kind,
            attributes=attributes,
            capture_args=capture_args,
            root=True,
        )

    return decorator

archipy.helpers.decorators.tracing.trace_class

trace_class(
    *,
    exclude: list[str] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[type], type]

Decorate a class so public methods are wrapped with OpenTelemetry spans.

Skips names starting with _, dunder methods, and property attributes. staticmethod and classmethod wrappers are preserved after wrapping the underlying function. Span names use ClassName.method_name.

Parameters:

Name	Type	Description	Default
exclude	list[str] | None	Method names to leave unwrapped.	None
capture_args	list[str] | None	Parameter names recorded as span attributes on all wrapped methods. Sensitive names (password, token, …) are redacted to ***.	None

Returns:

Type	Description
Callable[[type], type]	A class decorator.

Example

@trace_class(exclude=["helper"], capture_args=["user_id"])
class UserService:
    def get_user(self, user_id: int) -> dict[str, Any]:
        return {"id": user_id}

Source code in archipy/helpers/decorators/tracing.py

def trace_class(
    *,
    exclude: list[str] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[type], type]:
    """Decorate a class so public methods are wrapped with OpenTelemetry spans.

    Skips names starting with ``_``, dunder methods, and ``property`` attributes.
    ``staticmethod`` and ``classmethod`` wrappers are preserved after wrapping
    the underlying function. Span names use ``ClassName.method_name``.

    Args:
        exclude: Method names to leave unwrapped.
        capture_args: Parameter names recorded as span attributes on all wrapped methods.
            Sensitive names (password, token, …) are redacted to ``***``.

    Returns:
        A class decorator.

    Example:
        ```python
        @trace_class(exclude=["helper"], capture_args=["user_id"])
        class UserService:
            def get_user(self, user_id: int) -> dict[str, Any]:
                return {"id": user_id}
        ```
    """
    excluded = set(exclude or [])

    def decorator(cls: type) -> type:
        for method_name, attr in list(vars(cls).items()):
            if method_name.startswith("_") or method_name in excluded:
                continue
            if isinstance(attr, property | type):
                continue

            if isinstance(attr, staticmethod):
                wrapped = _wrap_method(cls, method_name, attr.__func__, capture_args=capture_args)
                setattr(cls, method_name, staticmethod(wrapped))
            elif isinstance(attr, classmethod):
                wrapped = _wrap_method(cls, method_name, attr.__func__, capture_args=capture_args)
                setattr(cls, method_name, classmethod(wrapped))
            elif callable(attr):
                setattr(cls, method_name, _wrap_method(cls, method_name, attr, capture_args=capture_args))

        return cls

    return decorator

options: show_root_toc_entry: false heading_level: 3

Metrics

OpenTelemetry metrics decorators for duration histograms and call counters (measure_duration, count_calls). Both accept sync and async functions; the async_* names are deprecated aliases that will be removed in 6.0.0.

OpenTelemetry metrics decorators for duration histograms and call counters.

archipy.helpers.decorators.metrics.logger module-attribute

logger = logging.getLogger(__name__)

archipy.helpers.decorators.metrics.clear_instrument_caches

clear_instrument_caches() -> None

Clear cached histogram and counter instruments.

Call after provider reset/re-init so instruments are rebound to the new MeterProvider. Used by OtelUtils.reset_for_testing().

Source code in archipy/helpers/decorators/metrics.py

def clear_instrument_caches() -> None:
    """Clear cached histogram and counter instruments.

    Call after provider reset/re-init so instruments are rebound to the new
    ``MeterProvider``. Used by ``OtelUtils.reset_for_testing()``.
    """
    with _CACHE_LOCK:
        _HISTOGRAM_CACHE.clear()
        _COUNTER_CACHE.clear()

archipy.helpers.decorators.metrics.measure_duration

measure_duration(
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _MetricDecorator

Decorate a sync or async function to record execution duration as a histogram.

Instrument name defaults to {module}.{qualname}.duration. The histogram is cached in a module-level map keyed by provider id + instrument name. Each recording includes a status attribute of ok, error or cancelled.

Works for def and async def functions alike. No-ops when OTel is disabled or config.OTEL.METRICS_ENABLED is False.

Parameters:

Name	Type	Description	Default
name	str | None	Histogram instrument name. Defaults to module-qualified duration name.	None
unit	str	Unit of the recorded duration. Defaults to "s".	's'
attributes	Mapping[str, AttributeValue] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
_MetricDecorator	A decorator that records duration around the target function.

Source code in archipy/helpers/decorators/metrics.py

def measure_duration(
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _MetricDecorator:
    """Decorate a sync or async function to record execution duration as a histogram.

    Instrument name defaults to ``{module}.{qualname}.duration``. The histogram
    is cached in a module-level map keyed by provider id + instrument name. Each
    recording includes a ``status`` attribute of ``ok``, ``error`` or ``cancelled``.

    Works for ``def`` and ``async def`` functions alike. No-ops when OTel is disabled
    or ``config.OTEL.METRICS_ENABLED`` is False.

    Args:
        name: Histogram instrument name. Defaults to module-qualified duration name.
        unit: Unit of the recorded duration. Defaults to ``"s"``.
        attributes: Optional static attributes merged into each recording.

    Returns:
        A decorator that records duration around the target function.
    """

    def decorator[**P, R](func: Callable[P, R]) -> Callable[P, R]:
        return _duration_wrapper(func, name, unit, attributes)

    return decorator

archipy.helpers.decorators.metrics.async_measure_duration

async_measure_duration(
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _AsyncMetricDecorator

Decorate an async function to record execution duration as a histogram.

Deprecated

Use :func:measure_duration, which supports async functions directly.

Parameters:

Name	Type	Description	Default
name	str | None	Histogram instrument name. Defaults to module-qualified duration name.	None
unit	str	Unit of the recorded duration. Defaults to "s".	's'
attributes	Mapping[str, AttributeValue] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
_AsyncMetricDecorator	A decorator that records duration around the target coroutine function.

Raises:

Type	Description
InvalidArgumentError	If the decorated object is not a coroutine function.

Source code in archipy/helpers/decorators/metrics.py

def async_measure_duration(
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _AsyncMetricDecorator:
    """Decorate an async function to record execution duration as a histogram.

    Deprecated:
        Use :func:`measure_duration`, which supports async functions directly.

    Args:
        name: Histogram instrument name. Defaults to module-qualified duration name.
        unit: Unit of the recorded duration. Defaults to ``"s"``.
        attributes: Optional static attributes merged into each recording.

    Returns:
        A decorator that records duration around the target coroutine function.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """
    warn_deprecated_name("async_measure_duration", "measure_duration")

    def decorator[**P, C: Coroutine[object, Never, object]](func: Callable[P, C]) -> Callable[P, C]:
        _require_coroutine_function(func, "async_measure_duration")
        return _duration_wrapper(func, name, unit, attributes)

    return decorator

archipy.helpers.decorators.metrics.count_calls

count_calls(
    name: str | None = None,
    *,
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _MetricDecorator

Decorate a sync or async function to increment a call counter.

Instrument name defaults to {module}.{qualname}.calls. The counter is cached in a module-level map keyed by provider id + instrument name. Each increment includes a status attribute of ok, error or cancelled.

Works for def and async def functions alike. No-ops when OTel is disabled or config.OTEL.METRICS_ENABLED is False.

Parameters:

Name	Type	Description	Default
name	str | None	Counter instrument name. Defaults to module-qualified calls name.	None
attributes	Mapping[str, AttributeValue] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
_MetricDecorator	A decorator that counts invocations of the target function.

Source code in archipy/helpers/decorators/metrics.py

def count_calls(
    name: str | None = None,
    *,
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _MetricDecorator:
    """Decorate a sync or async function to increment a call counter.

    Instrument name defaults to ``{module}.{qualname}.calls``. The counter is
    cached in a module-level map keyed by provider id + instrument name. Each
    increment includes a ``status`` attribute of ``ok``, ``error`` or ``cancelled``.

    Works for ``def`` and ``async def`` functions alike. No-ops when OTel is disabled
    or ``config.OTEL.METRICS_ENABLED`` is False.

    Args:
        name: Counter instrument name. Defaults to module-qualified calls name.
        attributes: Optional static attributes merged into each recording.

    Returns:
        A decorator that counts invocations of the target function.
    """

    def decorator[**P, R](func: Callable[P, R]) -> Callable[P, R]:
        return _counter_wrapper(func, name, attributes)

    return decorator

archipy.helpers.decorators.metrics.async_count_calls

async_count_calls(
    name: str | None = None,
    *,
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _AsyncMetricDecorator

Decorate an async function to increment a call counter.

Deprecated

Use :func:count_calls, which supports async functions directly.

Parameters:

Name	Type	Description	Default
name	str | None	Counter instrument name. Defaults to module-qualified calls name.	None
attributes	Mapping[str, AttributeValue] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
_AsyncMetricDecorator	A decorator that counts invocations of the target coroutine function.

Raises:

Type	Description
InvalidArgumentError	If the decorated object is not a coroutine function.

Source code in archipy/helpers/decorators/metrics.py

def async_count_calls(
    name: str | None = None,
    *,
    attributes: Mapping[str, AttributeValue] | None = None,
) -> _AsyncMetricDecorator:
    """Decorate an async function to increment a call counter.

    Deprecated:
        Use :func:`count_calls`, which supports async functions directly.

    Args:
        name: Counter instrument name. Defaults to module-qualified calls name.
        attributes: Optional static attributes merged into each recording.

    Returns:
        A decorator that counts invocations of the target coroutine function.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """
    warn_deprecated_name("async_count_calls", "count_calls")

    def decorator[**P, C: Coroutine[object, Never, object]](func: Callable[P, C]) -> Callable[P, C]:
        _require_coroutine_function(func, "async_count_calls")
        return _counter_wrapper(func, name, attributes)

    return decorator

options: show_root_toc_entry: false heading_level: 3

Singleton

Decorator that ensures a class is instantiated only once throughout the application lifecycle.

Singleton decorator helpers.

archipy.helpers.decorators.singleton_class.singleton_decorator module-attribute

singleton_decorator = deprecated_alias(
    "singleton_decorator", singleton
)

archipy.helpers.decorators.singleton_class.singleton

singleton(*, thread_safe: bool = True) -> Callable[[T], T]

A decorator to create thread-safe Singleton classes.

This decorator ensures that only one instance of a class is created. The decorated name is still the class itself, so isinstance, subclassing and class-level attributes keep working. Constructor arguments passed after the first instantiation are ignored.

Parameters:

Name	Type	Description	Default
thread_safe	bool	If True, enables thread-safety for instance creation. Defaults to True.	True

Returns:

Name	Type	Description
function	Callable[[T], T]	A decorator function that can be applied to a class.

Example

To create a Singleton class, apply the singleton and optionally specify whether thread-safety should be enabled:

@singleton(thread_safe=True)
class MySingletonClass:
    def __init__(self, value):
        self.value = value


# Create instances of MySingletonClass
instance1 = MySingletonClass(10)
instance2 = MySingletonClass(20)

# Verify that both instances are the same
print(instance1.value)  # Output: 10
print(instance2.value)  # Output: 10
print(instance1 is instance2)  # Output: True
print(isinstance(instance1, MySingletonClass))  # Output: True

Source code in archipy/helpers/decorators/singleton_class.py

def singleton[T: type[Any]](*, thread_safe: bool = True) -> Callable[[T], T]:  # Any: patches __new__/__init__
    """A decorator to create thread-safe Singleton classes.

    This decorator ensures that only one instance of a class is created. The decorated name
    is still the class itself, so ``isinstance``, subclassing and class-level attributes keep
    working. Constructor arguments passed after the first instantiation are ignored.

    Args:
        thread_safe (bool, optional): If True, enables thread-safety for instance creation.
                                      Defaults to True.

    Returns:
        function: A decorator function that can be applied to a class.

    Example:
        To create a Singleton class, apply the `singleton` and optionally specify
        whether thread-safety should be enabled:

        ```python
        @singleton(thread_safe=True)
        class MySingletonClass:
            def __init__(self, value):
                self.value = value


        # Create instances of MySingletonClass
        instance1 = MySingletonClass(10)
        instance2 = MySingletonClass(20)

        # Verify that both instances are the same
        print(instance1.value)  # Output: 10
        print(instance2.value)  # Output: 10
        print(instance1 is instance2)  # Output: True
        print(isinstance(instance1, MySingletonClass))  # Output: True
        ```
    """

    def decorator(cls: T) -> T:
        """Patch the class so that it only ever builds and initializes one instance.

        Args:
            cls: The class to be decorated as a Singleton.

        Returns:
            The same class, now limited to a single instance per (sub)class.
        """
        instances: dict[type, object] = {}
        initialized: set[type] = set()
        # Without thread safety a no-op context manager replaces the lock
        lock: AbstractContextManager[object] = threading.RLock() if thread_safe else nullcontext()
        original_new: Callable[..., object] = cls.__new__
        original_init: Callable[..., None] = cls.__init__

        def singleton_new(klass: type, *args: object, **kwargs: object) -> object:
            if klass not in instances:
                with lock:
                    if klass not in instances:
                        if original_new is object.__new__:
                            instances[klass] = original_new(klass)
                        else:
                            instances[klass] = original_new(klass, *args, **kwargs)
            return instances[klass]

        def singleton_init(self: object, *args: object, **kwargs: object) -> None:
            klass = type(self)
            if klass not in initialized:
                with lock:
                    if klass not in initialized:
                        original_init(self, *args, **kwargs)
                        initialized.add(klass)

        cls.__new__ = singleton_new  # type: ignore[assignment]
        cls.__init__ = singleton_init  # type: ignore[misc]
        return cls

    return decorator

options: show_root_toc_entry: false heading_level: 3

SQLAlchemy Atomic

The SQLAlchemy atomic decorators moved to the adapters layer; see Base SQLAlchemy. The old archipy.helpers.decorators names still work and emit a DeprecationWarning; they will be removed in 6.0.0.

Deprecation Warnings

Decorator that emits a deprecation warning when a decorated function or class is used.

Decorators that warn on deprecated usage.

archipy.helpers.decorators.deprecation_warnings.method_deprecation_warning

method_deprecation_warning(
    message: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

A decorator that issues a deprecation warning when the decorated method is called.

Parameters:

Name	Type	Description	Default
message	str	The deprecation message to display when the method is called. Defaults to "This method is deprecated and will be removed in a future version."	None

Returns:

Name	Type	Description
Callable	Callable[[Callable[P, R]], Callable[P, R]]	The decorated method that issues a deprecation warning.

Example

To use this decorator, apply it to a method:

class MyClass:
    @method_deprecation_warning("This method is deprecated and will be removed in a future version.")
    def old_method(self):
        return "This is the old method."


# Calling the method will issue a deprecation warning
obj = MyClass()
result = obj.old_method()

Output:

DeprecationWarning: This method is deprecated and will be removed in a future version.
This is the old method.

Source code in archipy/helpers/decorators/deprecation_warnings.py

def method_deprecation_warning[**P, R](message: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator that issues a deprecation warning when the decorated method is called.

    Args:
        message (str, optional): The deprecation message to display when the method is called.
            Defaults to "This method is deprecated and will be removed in a future version."

    Returns:
        Callable: The decorated method that issues a deprecation warning.

    Example:
        To use this decorator, apply it to a method:

        ```python
        class MyClass:
            @method_deprecation_warning("This method is deprecated and will be removed in a future version.")
            def old_method(self):
                return "This is the old method."


        # Calling the method will issue a deprecation warning
        obj = MyClass()
        result = obj.old_method()
        ```

        Output:
        ```
        DeprecationWarning: This method is deprecated and will be removed in a future version.
        This is the old method.
        ```
    """
    default_message = "This method is deprecated and will be removed in a future version."
    final_message = message if message is not None else default_message

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            warnings.warn(final_message, DeprecationWarning, stacklevel=2)
            return func(*args, **kwargs)

        return wrapper

    return decorator

archipy.helpers.decorators.deprecation_warnings.class_deprecation_warning

class_deprecation_warning(
    message: str | None = None,
) -> Callable[[T], T]

A decorator that issues a deprecation warning when the decorated class is instantiated.

Parameters:

Name	Type	Description	Default
message	str	The deprecation message to display when the class is instantiated. Defaults to "This class is deprecated and will be removed in a future version."	None

Returns:

Name	Type	Description
Callable	Callable[[T], T]	The decorated class that issues a deprecation warning.

Example

To use this decorator, apply it to a class:

@class_deprecation_warning("This class is deprecated and will be removed in a future version.")
class OldClass:
    def __init__(self):
        pass


# Instantiating the class will issue a deprecation warning
obj = OldClass()

Output:

DeprecationWarning: This class is deprecated and will be removed in a future version.

Source code in archipy/helpers/decorators/deprecation_warnings.py

def class_deprecation_warning[T: type[Any]](message: str | None = None) -> Callable[[T], T]:  # Any: patches __init__
    """A decorator that issues a deprecation warning when the decorated class is instantiated.

    Args:
        message (str, optional): The deprecation message to display when the class is instantiated.
            Defaults to "This class is deprecated and will be removed in a future version."

    Returns:
        Callable: The decorated class that issues a deprecation warning.

    Example:
        To use this decorator, apply it to a class:

        ```python
        @class_deprecation_warning("This class is deprecated and will be removed in a future version.")
        class OldClass:
            def __init__(self):
                pass


        # Instantiating the class will issue a deprecation warning
        obj = OldClass()
        ```

        Output:
        ```
        DeprecationWarning: This class is deprecated and will be removed in a future version.
        ```
    """
    default_message = "This class is deprecated and will be removed in a future version."
    final_message = message if message is not None else default_message

    def decorator(cls: T) -> T:
        original_init: Callable[..., None] = cls.__init__

        @wraps(original_init)
        def new_init(self: object, *args: object, **kwargs: object) -> None:
            warnings.warn(final_message, DeprecationWarning, stacklevel=2)
            original_init(self, *args, **kwargs)

        cls.__init__ = new_init
        return cls

    return decorator

options: show_root_toc_entry: false heading_level: 3

Deprecation Exception

Decorator that raises an exception when a deprecated function or class is called.

Decorators that raise on deprecated usage.

archipy.helpers.decorators.deprecation_exception.method_deprecation_error

method_deprecation_error(
    operation: str | None = None,
    lang: LanguageType = LanguageType.EN,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Decorator that raises a DeprecationError when the decorated method is called.

This decorator is used to mark methods as deprecated and immediately prevent their use by raising a DeprecationError when they are called. This is stricter than a warning and ensures deprecated methods cannot be used.

Parameters:

Name	Type	Description	Default
operation	str	The name of the operation that is deprecated. Defaults to the name of the decorated method.	None
lang	LanguageType	The language for the error message (default: "en").	EN

Returns:

Name	Type	Description
Callable	Callable[[Callable[P, R]], Callable[P, R]]	The decorated method that raises a DeprecationException.

Example

To use this decorator, apply it to a method:

class MyClass:
    @method_deprecation_error(operation="old_method", lang=LanguageType.EN)
    def old_method(self):
        return "This is the old method."


# Calling the method will raise a DeprecationException
obj = MyClass()
result = obj.old_method()

Output:

DeprecationException: This operation is deprecated and will be removed in a future version.
Operation: old_method

Source code in archipy/helpers/decorators/deprecation_exception.py

def method_deprecation_error[**P, R](
    operation: str | None = None,
    lang: LanguageType = LanguageType.EN,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator that raises a DeprecationError when the decorated method is called.

    This decorator is used to mark methods as deprecated and immediately prevent
    their use by raising a DeprecationError when they are called. This is stricter
    than a warning and ensures deprecated methods cannot be used.

    Args:
        operation (str, optional): The name of the operation that is deprecated.
            Defaults to the name of the decorated method.
        lang (LanguageType): The language for the error message (default: "en").

    Returns:
        Callable: The decorated method that raises a DeprecationException.

    Example:
        To use this decorator, apply it to a method:

        ```python
        class MyClass:
            @method_deprecation_error(operation="old_method", lang=LanguageType.EN)
            def old_method(self):
                return "This is the old method."


        # Calling the method will raise a DeprecationException
        obj = MyClass()
        result = obj.old_method()
        ```

        Output:
        ```
        DeprecationException: This operation is deprecated and will be removed in a future version.
        Operation: old_method
        ```
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        # Capture function name before wrapping - use getattr for type safety
        func_name = getattr(func, "__name__", "unknown")

        @wraps(func)
        def wrapper(*_args: P.args, **_kwargs: P.kwargs) -> R:
            operation_name = operation if operation is not None else func_name
            raise DeprecationError(deprecated_feature=operation_name, lang=lang)

        return wrapper

    return decorator

archipy.helpers.decorators.deprecation_exception.class_deprecation_error

class_deprecation_error(
    operation: str | None = None,
    lang: LanguageType = LanguageType.FA,
) -> Callable[[T], T]

A decorator that raises a DeprecationException when the decorated class is instantiated.

Parameters:

Name	Type	Description	Default
operation	str	The name of the operation that is deprecated. Defaults to the name of the decorated class.	None
lang	str	The language for the error message (default: "fa").	FA

Returns:

Name	Type	Description
Callable	Callable[[T], T]	The decorated class that raises a DeprecationException.

Example

To use this decorator, apply it to a class:

@class_deprecation_error(operation="OldClass", lang="en")
class OldClass:
    def __init__(self):
        pass


# Instantiating the class will raise a DeprecationException
obj = OldClass()

Output:

DeprecationException: This operation is deprecated and will be removed in a future version.
Operation: OldClass

Source code in archipy/helpers/decorators/deprecation_exception.py

def class_deprecation_error[T: type[Any]](  # Any: patches __init__ on the class
    operation: str | None = None,
    lang: LanguageType = LanguageType.FA,
) -> Callable[[T], T]:
    """A decorator that raises a DeprecationException when the decorated class is instantiated.

    Args:
        operation (str, optional): The name of the operation that is deprecated.
            Defaults to the name of the decorated class.
        lang (str): The language for the error message (default: "fa").

    Returns:
        Callable: The decorated class that raises a DeprecationException.

    Example:
        To use this decorator, apply it to a class:

        ```python
        @class_deprecation_error(operation="OldClass", lang="en")
        class OldClass:
            def __init__(self):
                pass


        # Instantiating the class will raise a DeprecationException
        obj = OldClass()
        ```

        Output:
        ```
        DeprecationException: This operation is deprecated and will be removed in a future version.
        Operation: OldClass
        ```
    """

    def decorator(cls: T) -> T:
        @wraps(cls.__init__)
        def new_init(_self: object, *_args: object, **_kwargs: object) -> None:
            operation_name = operation if operation is not None else cls.__name__
            raise DeprecationError(deprecated_feature=operation_name, lang=lang)

        cls.__init__ = new_init
        return cls

    return decorator

options: show_root_toc_entry: false heading_level: 3

gRPC Rate Limit

Decorator that declares per-RPC rate-limit windows on gRPC servicer methods. Read by GrpcServerRateLimitInterceptor and AsyncGrpcServerRateLimitInterceptor.

Decorator for declaring per-RPC gRPC rate-limit windows on servicer methods.

archipy.helpers.decorators.grpc_rate_limit_window.grpc_rate_limit_decorator module-attribute

grpc_rate_limit_decorator = deprecated_alias(
    "grpc_rate_limit_decorator", grpc_rate_limit
)

archipy.helpers.decorators.grpc_rate_limit_window.grpc_rate_limit

grpc_rate_limit(
    calls_count: int = 1,
    milliseconds: int = 0,
    seconds: int = 0,
    minutes: int = 0,
    hours: int = 0,
    days: int = 0,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Attach a rate-limit window to a gRPC servicer method without wrapping it.

Stack multiple decorators on one method to enforce burst and sustained limits. The GrpcServerRateLimitInterceptor reads __archipy_rate_limit_windows__ and enforces each window in declaration order.

Parameters:

Name	Type	Description	Default
calls_count	int	Maximum allowed requests within the window.	1
milliseconds	int	Milliseconds component of the window.	0
seconds	int	Seconds component of the window.	0
minutes	int	Minutes component of the window.	0
hours	int	Hours component of the window.	0
days	int	Days component of the window.	0

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	A decorator that records the window on the target function.

Raises:

Type	Description
InvalidArgumentError	If calls_count or the computed window is invalid.

Source code in archipy/helpers/decorators/grpc_rate_limit_window.py

def grpc_rate_limit[**P, R](
    calls_count: int = 1,
    milliseconds: int = 0,
    seconds: int = 0,
    minutes: int = 0,
    hours: int = 0,
    days: int = 0,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Attach a rate-limit window to a gRPC servicer method without wrapping it.

    Stack multiple decorators on one method to enforce burst and sustained limits.
    The ``GrpcServerRateLimitInterceptor`` reads ``__archipy_rate_limit_windows__``
    and enforces each window in declaration order.

    Args:
        calls_count: Maximum allowed requests within the window.
        milliseconds: Milliseconds component of the window.
        seconds: Seconds component of the window.
        minutes: Minutes component of the window.
        hours: Hours component of the window.
        days: Days component of the window.

    Returns:
        A decorator that records the window on the target function.

    Raises:
        InvalidArgumentError: If ``calls_count`` or the computed window is invalid.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        window = RateLimitUtils.compute_rate_limit_window(
            calls_count=calls_count,
            milliseconds=milliseconds,
            seconds=seconds,
            minutes=minutes,
            hours=hours,
            days=days,
        )
        existing = getattr(func, RateLimitUtils.ARCHIPY_RATE_LIMIT_WINDOWS_ATTR, ())
        setattr(func, RateLimitUtils.ARCHIPY_RATE_LIMIT_WINDOWS_ATTR, (*existing, window))
        return func

    return decorator

options: show_root_toc_entry: false heading_level: 3