Decorators

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

Cache

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

Caching decorator helpers.

archipy.helpers.decorators.cache.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_decorator(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/cache.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_decorator(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: Any, instance: object | 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).
        """
        self._func: Callable[..., R] = func
        self._cache: Any = cache
        self._instance: object | None = instance
        # Preserve function metadata
        wraps(func)(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")
        bound_method_name = f"_cached_{func_name}"
        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)
            # 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)

        return getattr(obj, bound_method_name)

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

        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).
        """
        # Create a key based on function name, args, and kwargs
        func_name = getattr(self._func, "__name__", "unknown")
        key_parts = [func_name]

        # Use repr() with type information for robust key generation
        key_parts.extend(f"{type(arg).__name__}:{arg!r}" for arg in args)

        # Add keyword arguments
        key_parts.extend(f"{k}={type(v).__name__}:{v!r}" for k, v in sorted(kwargs.items()))

        key = ":".join(key_parts)

        # Check if result is in cache
        if key in self._cache:
            return self._cache[key]

        # Call the function with the instance if this is a bound method
        if self._instance is not None:
            result: R = self._func(self._instance, *args, **kwargs)
        else:
            result = self._func(*args, **kwargs)

        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.
        """
        self._cache.clear()

archipy.helpers.decorators.cache.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/cache.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.
    """
    self._cache.clear()

archipy.helpers.decorators.cache.ttl_cache_decorator

ttl_cache_decorator(
    ttl_seconds: int | Callable[[], int] = 300,
    maxsize: int = 100,
) -> Any

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
Any	Decorated function with TTL caching and a clear_cache() method (typed as :class:~typing.Any
Any	at compile time so ParamSpec-style decorator typing stays sound).

Example

class DataService:
    @ttl_cache_decorator(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
• Cache is shared across all instances of a class (not per-instance)

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

def ttl_cache_decorator[**P, R](
    ttl_seconds: int | Callable[[], int] = 300,
    maxsize: int = 100,
) -> Any:
    """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 (typed as :class:`~typing.Any`
        at compile time so ParamSpec-style decorator typing stays sound).

    Example:
        ```python
        class DataService:
            @ttl_cache_decorator(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
        - Cache is shared across all instances of a class (not per-instance)
    """
    from cachetools import TTLCache

    resolved_ttl = int(ttl_seconds) if isinstance(ttl_seconds, int) else int(ttl_seconds())
    cache: TTLCache = TTLCache(maxsize=maxsize, ttl=resolved_ttl)

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

    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.P module-attribute

P = ParamSpec('P')

archipy.helpers.decorators.retry.R module-attribute

R = TypeVar('R')

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

logger = logging.getLogger(__name__)

archipy.helpers.decorators.retry.retry_decorator

retry_decorator(
    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,
) -> Callable[[Callable[P, R]], Callable[P, R]]

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

Parameters:

Name	Type	Description	Default
max_retries	int	The maximum number of retry attempts. Defaults to 3.	3
delay	float	The delay (in seconds) between retries. 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. Defaults to None.	None
resource_type	Optional[str]	The type of resource being exhausted. Defaults to None.	None

Returns:

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

Example

To use this decorator, apply it to a function:

@retry_decorator(max_retries=3, delay=1, retry_on=(ValueError,), ignore=(TypeError,), 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.py

def retry_decorator(
    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,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator that retries a function when it raises an exception.

    Args:
        max_retries (int): The maximum number of retry attempts. Defaults to 3.
        delay (float): The delay (in seconds) between retries. 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. Defaults to None.
        resource_type (Optional[str]): The type of resource being exhausted. Defaults to None.

    Returns:
        Callable: The decorated function with retry logic.

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

        ```python
        @retry_decorator(max_retries=3, delay=1, retry_on=(ValueError,), ignore=(TypeError,), 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
        ```
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        from functools import wraps

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            retries = 0
            while retries < max_retries:
                try:
                    result = func(*args, **kwargs)
                    if retries > 0:
                        logger.info("Attempt %d succeeded.", retries + 1)
                except Exception as e:
                    retries += 1
                    # Check if the exception should be ignored
                    if ignore and isinstance(e, ignore):
                        raise
                    # Check if the exception should be retried
                    if retry_on and not isinstance(e, retry_on):
                        raise
                    logger.warning("Attempt %d failed: %s", retries, e)
                    if retries < max_retries:
                        time.sleep(delay)
                    continue
                return result
            raise ResourceExhaustedError(resource_type=resource_type)

        return wrapper

    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.P module-attribute

P = ParamSpec('P')

archipy.helpers.decorators.timeout.R module-attribute

R = TypeVar('R')

archipy.helpers.decorators.timeout.timeout_decorator

timeout_decorator(
    seconds: int,
) -> 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 DeadlineExceededException is raised.

Parameters:

Name	Type	Description	Default
seconds	int	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_decorator(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 DeadlineExceededException as e:
    print(e)  # Output: "Function long_running_function timed out after 3 seconds."

Output:

DeadlineExceededException: Function long_running_function timed out after 3 seconds.

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

def timeout_decorator(seconds: int) -> 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 DeadlineExceededException is raised.

    Args:
        seconds (int): 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_decorator(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 DeadlineExceededException as e:
            print(e)  # Output: "Function long_running_function timed out after 3 seconds."
        ```

        Output:
        ```
        DeadlineExceededException: Function long_running_function timed out after 3 seconds.
        ```
    """

    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:
            def handle_timeout(_signum: int, _frame: Any) -> None:
                raise DeadlineExceededError(operation=func_name)

            # Set the signal handler and alarm
            signal.signal(signal.SIGALRM, handle_timeout)
            signal.alarm(seconds)

            try:
                result = func(*args, **kwargs)
            finally:
                # Disable the alarm
                signal.alarm(0)
            return result

        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.logger module-attribute

logger = logging.getLogger(__name__)

archipy.helpers.decorators.timing.timing_decorator

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

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

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 if the logging level is DEBUG.

Example

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

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


result = example_function(2)

Output (if logging level is DEBUG):

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.py

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

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

    Returns:
        Callable: The wrapped function which logs the execution time if the logging level is DEBUG.

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

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


        result = example_function(2)
        ```

        Output (if logging level is DEBUG):
        ```
        2023-10-10 12:00:00,000 - DEBUG - example_function took 2.0001 seconds to execute.
        Slept for 2 seconds
        ```
    """
    from functools import wraps

    # 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:
        if logging.getLogger().level == logging.DEBUG:
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            logger.debug("%s took %.4f seconds to execute.", func_name, end_time - start_time)
        else:
            result = func(*args, **kwargs)
        return result

    return wrapper

options: show_root_toc_entry: false heading_level: 3

Tracing

OpenTelemetry span decorators for functions and classes (trace_span, async_trace_span, trace_root, async_trace_root, trace_class).

OpenTelemetry tracing decorators for function and class instrumentation.

archipy.helpers.decorators.tracing.trace_span

trace_span(
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorate a sync function with an OpenTelemetry child span.

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	Any | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	dict[str, Any] | 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
Callable[[F], Callable[..., Any]]	A decorator that wraps the target function in a span.

Raises:

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

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[F: _Function](
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorate a sync 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.

    Raises:
        InvalidArgumentError: If the decorated object is a coroutine function.

    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(func: F) -> Callable[..., Any]:
        _reject_coroutine_function(func, "trace_span")
        span_name = name or func.__name__
        signature = inspect.signature(func)

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return _run_traced(
                func,
                args,
                kwargs,
                span_name=span_name,
                kind=kind,
                attributes=attributes,
                capture_args=capture_args,
                signature=signature,
                root=False,
            )

        return wrapper

    return decorator

archipy.helpers.decorators.tracing.async_trace_span

async_trace_span(
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]

Decorate an async function with an OpenTelemetry child span.

Parameters:

Name	Type	Description	Default
name	str | None	Span name. Defaults to the function name.	None
kind	Any | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	dict[str, Any] | 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
Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]	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[F: _AsyncFunction](
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]:
    """Decorate an 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 coroutine function in a span.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """

    def decorator(func: F) -> Callable[..., Coroutine[Any, Any, Any]]:
        if not inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "async_trace_span",
                    "func_name": func.__name__,
                },
            )

        span_name = name or func.__name__
        signature = inspect.signature(func)

        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            return await _run_traced_async(
                func,
                args,
                kwargs,
                span_name=span_name,
                kind=kind,
                attributes=attributes,
                capture_args=capture_args,
                signature=signature,
                root=False,
            )

        return wrapper

    return decorator

archipy.helpers.decorators.tracing.trace_root

trace_root(
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorate a sync 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	Any | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	dict[str, Any] | 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
Callable[[F], Callable[..., Any]]	A decorator that wraps the target function in a root span.

Raises:

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

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

def trace_root[F: _Function](
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorate a sync 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.

    Raises:
        InvalidArgumentError: If the decorated object is a coroutine function.
    """

    def decorator(func: F) -> Callable[..., Any]:
        _reject_coroutine_function(func, "trace_root")
        span_name = name or func.__name__
        signature = inspect.signature(func)

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return _run_traced(
                func,
                args,
                kwargs,
                span_name=span_name,
                kind=kind,
                attributes=attributes,
                capture_args=capture_args,
                signature=signature,
                root=True,
            )

        return wrapper

    return decorator

archipy.helpers.decorators.tracing.async_trace_root

async_trace_root(
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]

Decorate an 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	Any | None	OpenTelemetry SpanKind. Defaults to SpanKind.INTERNAL.	None
attributes	dict[str, Any] | 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
Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]	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[F: _AsyncFunction](
    name: str | None = None,
    *,
    kind: Any | None = None,
    attributes: dict[str, Any] | None = None,
    capture_args: list[str] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]:
    """Decorate an 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 coroutine function in a root span.

    Raises:
        InvalidArgumentError: If the decorated object is not a coroutine function.
    """

    def decorator(func: F) -> Callable[..., Coroutine[Any, Any, Any]]:
        if not inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "async_trace_root",
                    "func_name": func.__name__,
                },
            )

        span_name = name or func.__name__
        signature = inspect.signature(func)

        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            return await _run_traced_async(
                func,
                args,
                kwargs,
                span_name=span_name,
                kind=kind,
                attributes=attributes,
                capture_args=capture_args,
                signature=signature,
                root=True,
            )

        return wrapper

    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):
                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, async_measure_duration, count_calls, async_count_calls).

OpenTelemetry metrics decorators for duration histograms and call counters.

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: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorate a sync 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 or error.

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	dict[str, Any] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
Callable[[F], Callable[..., Any]]	A decorator that records duration around the target function.

Raises:

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

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

def measure_duration[F: _Function](
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorate a sync 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`` or ``error``.

    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.

    Raises:
        InvalidArgumentError: If the decorated object is a coroutine function.
    """

    def decorator(func: F) -> Callable[..., Any]:
        if inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "measure_duration",
                    "func_name": func.__name__,
                    "hint": "Use async_measure_duration instead of measure_duration",
                },
            )

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            config = BaseConfig.global_config()
            if not _metrics_active(config):
                return func(*args, **kwargs)

            OtelUtils.init_otel_if_needed(config)
            if OtelUtils.import_failed() or not OtelUtils.is_metrics_enabled(config):
                return func(*args, **kwargs)

            histogram = _get_or_create_histogram(func, name, unit)
            start = time.perf_counter()
            status = "ok"
            try:
                return func(*args, **kwargs)
            except Exception as exc:
                status = OtelUtils.metric_status_for_exception(exc)
                raise
            finally:
                histogram.record(
                    time.perf_counter() - start,
                    _merge_status_attributes(attributes, status),
                )

        return wrapper

    return decorator

archipy.helpers.decorators.metrics.async_measure_duration

async_measure_duration(
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]

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

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	dict[str, Any] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]	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[F: _AsyncFunction](
    name: str | None = None,
    *,
    unit: str = "s",
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]:
    """Decorate an async function to record execution duration as a histogram.

    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.
    """

    def decorator(func: F) -> Callable[..., Coroutine[Any, Any, Any]]:
        if not inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "async_measure_duration",
                    "func_name": func.__name__,
                },
            )

        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            config = BaseConfig.global_config()
            if not _metrics_active(config):
                return await func(*args, **kwargs)

            OtelUtils.init_otel_if_needed(config)
            if OtelUtils.import_failed() or not OtelUtils.is_metrics_enabled(config):
                return await func(*args, **kwargs)

            histogram = _get_or_create_histogram(func, name, unit)
            start = time.perf_counter()
            status = "ok"
            try:
                return await func(*args, **kwargs)
            except asyncio.CancelledError:
                status = "cancelled"
                raise
            except Exception as exc:
                status = OtelUtils.metric_status_for_exception(exc)
                raise
            finally:
                histogram.record(
                    time.perf_counter() - start,
                    _merge_status_attributes(attributes, status),
                )

        return wrapper

    return decorator

archipy.helpers.decorators.metrics.count_calls

count_calls(
    name: str | None = None,
    *,
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorate a sync 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 or error.

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	dict[str, Any] | None	Optional static attributes merged into each recording.	None

Returns:

Type	Description
Callable[[F], Callable[..., Any]]	A decorator that counts invocations of the target function.

Raises:

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

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

def count_calls[F: _Function](
    name: str | None = None,
    *,
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorate a sync 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`` or ``error``.

    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.

    Raises:
        InvalidArgumentError: If the decorated object is a coroutine function.
    """

    def decorator(func: F) -> Callable[..., Any]:
        if inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "count_calls",
                    "func_name": func.__name__,
                    "hint": "Use async_count_calls instead of count_calls",
                },
            )

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            config = BaseConfig.global_config()
            if not _metrics_active(config):
                return func(*args, **kwargs)

            OtelUtils.init_otel_if_needed(config)
            if OtelUtils.import_failed() or not OtelUtils.is_metrics_enabled(config):
                return func(*args, **kwargs)

            counter = _get_or_create_counter(func, name)
            status = "ok"
            try:
                return func(*args, **kwargs)
            except Exception as exc:
                status = OtelUtils.metric_status_for_exception(exc)
                raise
            finally:
                counter.add(1, _merge_status_attributes(attributes, status))

        return wrapper

    return decorator

archipy.helpers.decorators.metrics.async_count_calls

async_count_calls(
    name: str | None = None,
    *,
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]

Decorate an async function to increment a call counter.

Parameters:

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

Returns:

Type	Description
Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]	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[F: _AsyncFunction](
    name: str | None = None,
    *,
    attributes: dict[str, Any] | None = None,
) -> Callable[[F], Callable[..., Coroutine[Any, Any, Any]]]:
    """Decorate an async function to increment a call counter.

    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.
    """

    def decorator(func: F) -> Callable[..., Coroutine[Any, Any, Any]]:
        if not inspect.iscoroutinefunction(func):
            raise InvalidArgumentError(
                argument_name="func",
                additional_data={
                    "decorator": "async_count_calls",
                    "func_name": func.__name__,
                },
            )

        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            config = BaseConfig.global_config()
            if not _metrics_active(config):
                return await func(*args, **kwargs)

            OtelUtils.init_otel_if_needed(config)
            if OtelUtils.import_failed() or not OtelUtils.is_metrics_enabled(config):
                return await func(*args, **kwargs)

            counter = _get_or_create_counter(func, name)
            status = "ok"
            try:
                return await func(*args, **kwargs)
            except asyncio.CancelledError:
                status = "cancelled"
                raise
            except Exception as exc:
                status = OtelUtils.metric_status_for_exception(exc)
                raise
            finally:
                counter.add(1, _merge_status_attributes(attributes, status))

        return wrapper

    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.singleton_decorator

singleton_decorator(
    *, thread_safe: bool = True
) -> Callable[[type[Any]], Callable[..., Any]]

A decorator to create thread-safe Singleton classes.

This decorator ensures that only one instance of a class is created. It supports an optional thread_safe parameter to control whether thread-safety mechanisms (e.g., locks) should be used.

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[[type[Any]], Callable[..., Any]]	A decorator function that can be applied to a class.

Example

To create a Singleton class, apply the singleton decorator 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

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

def singleton_decorator(*, thread_safe: bool = True) -> Callable[[type[Any]], Callable[..., Any]]:
    """A decorator to create thread-safe Singleton classes.

    This decorator ensures that only one instance of a class is created. It supports an optional
    `thread_safe` parameter to control whether thread-safety mechanisms (e.g., locks) should be used.

    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` decorator 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
        ```
    """

    def decorator(cls: type[Any]) -> Callable[..., Any]:
        """The inner decorator function that implements the Singleton pattern.

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

        Returns:
            function: A function that returns the Singleton instance of the class.
        """
        instances = {}  # Stores instances of Singleton classes
        lock: threading.Lock | None = (
            threading.Lock() if thread_safe else None
        )  # Lock for thread-safe instance creation

        def get_instance(*args: Any, **kwargs: Any) -> Any:
            """Create or return the Singleton instance of the class.

            If `thread_safe` is True, a lock is used to ensure that only one instance is created
            even in a multi-threaded environment. If `thread_safe` is False, no locking mechanism
            is used, which may result in multiple instances being created in a multi-threaded context.

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

            Returns:
                object: The Singleton instance of the class.
            """
            if cls not in instances:
                if thread_safe:
                    if lock is not None:
                        with lock:
                            if cls not in instances:
                                instances[cls] = cls(*args, **kwargs)
                else:
                    instances[cls] = cls(*args, **kwargs)
            return instances[cls]

        return get_instance

    return decorator

options: show_root_toc_entry: false heading_level: 3

SQLAlchemy Atomic

Decorator that wraps a function in a SQLAlchemy database transaction, rolling back on failure.

SQLAlchemy atomic transaction decorators.

This module provides decorators for managing SQLAlchemy transactions with automatic commit/rollback and support for different database types (PostgreSQL, SQLite, StarRocks, MySQL).

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

logger = logging.getLogger(__name__)

archipy.helpers.decorators.sqlalchemy_atomic.ATOMIC_BLOCK_CONFIGS module-attribute

ATOMIC_BLOCK_CONFIGS = {
    "postgres": {
        "flag": "in_postgres_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.postgres.sqlalchemy.session_manager_registry.PostgresSessionManagerRegistry",
    },
    "sqlite": {
        "flag": "in_sqlite_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.sqlite.sqlalchemy.session_manager_registry.SQLiteSessionManagerRegistry",
    },
    "starrocks": {
        "flag": "in_starrocks_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry",
    },
    "mysql": {
        "flag": "in_mysql_sqlalchemy_atomic_block",
        "registry": "archipy.adapters.mysql.sqlalchemy.session_manager_registry.MySQLSessionManagerRegistry",
    },
}

archipy.helpers.decorators.sqlalchemy_atomic.R module-attribute

R = TypeVar('R')

archipy.helpers.decorators.sqlalchemy_atomic.sqlalchemy_atomic_decorator

sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: Literal[False] = False,
    function: Callable[..., R] = ...,
) -> Callable[..., R]

sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: Literal[True],
    function: Callable[..., Awaitable[R]] = ...,
) -> Callable[..., Awaitable[R]]

sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: bool = False,
    function: None = None,
) -> partial[Callable[..., Any]]

sqlalchemy_atomic_decorator(
    db_type: str,
    is_async: bool = False,
    function: Callable[..., R]
    | Callable[..., Awaitable[R]]
    | None = None,
) -> (
    Callable[..., R]
    | Callable[..., Awaitable[R]]
    | partial[Callable[..., Any]]
)

Factory for creating SQLAlchemy atomic transaction decorators.

This decorator ensures that a function runs within a database transaction for the specified database type. If the function succeeds, the transaction is committed; otherwise, it is rolled back. Supports both synchronous and asynchronous functions.

Parameters:

Name	Type	Description	Default
db_type	str	The database type ("postgres", "sqlite", "starrocks", or "mysql").	required
is_async	bool	Whether the function is asynchronous. Defaults to False.	False
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., R] | Callable[..., Awaitable[R]] | partial[Callable[..., Any]]	Callable | partial: The wrapped function or a partial function for later use.

Raises:

Type	Description
ValueError	If an invalid db_type is provided.
DatabaseSerializationError	If a serialization failure is detected.
DatabaseDeadlockError	If an operational error occurs due to a deadlock.
DatabaseTransactionError	If a transaction-related error occurs.
DatabaseQueryError	If a query-related error occurs.
DatabaseConnectionError	If a connection-related error occurs.
DatabaseConstraintError	If a constraint violation is detected.
DatabaseIntegrityError	If an integrity violation is detected.
DatabaseTimeoutError	If a database operation times out.
DatabaseConfigurationError	If there's an error in the database configuration.

Example

Synchronous PostgreSQL example

@sqlalchemy_atomic_decorator(db_type="postgres") def update_user(id: int, name: str) -> None: # Database operations pass

Asynchronous SQLite example

@sqlalchemy_atomic_decorator(db_type="sqlite", is_async=True) async def update_record(id: int, data: str) -> None: # Async database operations pass

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

def sqlalchemy_atomic_decorator[R](
    db_type: str,
    is_async: bool = False,
    function: Callable[..., R] | Callable[..., Awaitable[R]] | None = None,
) -> Callable[..., R] | Callable[..., Awaitable[R]] | partial[Callable[..., Any]]:
    """Factory for creating SQLAlchemy atomic transaction decorators.

    This decorator ensures that a function runs within a database transaction for the specified
    database type. If the function succeeds, the transaction is committed; otherwise, it is rolled back.
    Supports both synchronous and asynchronous functions.

    Args:
        db_type (str): The database type ("postgres", "sqlite", "starrocks", or "mysql").
        is_async (bool): Whether the function is asynchronous. Defaults to False.
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.

    Raises:
        ValueError: If an invalid db_type is provided.
        DatabaseSerializationError: If a serialization failure is detected.
        DatabaseDeadlockError: If an operational error occurs due to a deadlock.
        DatabaseTransactionError: If a transaction-related error occurs.
        DatabaseQueryError: If a query-related error occurs.
        DatabaseConnectionError: If a connection-related error occurs.
        DatabaseConstraintError: If a constraint violation is detected.
        DatabaseIntegrityError: If an integrity violation is detected.
        DatabaseTimeoutError: If a database operation times out.
        DatabaseConfigurationError: If there's an error in the database configuration.

    Example:
        # Synchronous PostgreSQL example
        @sqlalchemy_atomic_decorator(db_type="postgres")
        def update_user(id: int, name: str) -> None:
            # Database operations
            pass

        # Asynchronous SQLite example
        @sqlalchemy_atomic_decorator(db_type="sqlite", is_async=True)
        async def update_record(id: int, data: str) -> None:
            # Async database operations
            pass
    """
    if db_type not in ATOMIC_BLOCK_CONFIGS:
        raise InvalidArgumentError(
            argument_name="db_type",
            additional_data={"valid_values": list(ATOMIC_BLOCK_CONFIGS.keys())},
        )

    atomic_flag = ATOMIC_BLOCK_CONFIGS[db_type]["flag"]

    if is_async:
        if function is not None:
            return _create_async_atomic_wrapper(
                cast("Callable[..., Awaitable[R]]", function),
                db_type=db_type,
                atomic_flag=atomic_flag,
            )
        return partial(sqlalchemy_atomic_decorator, db_type=db_type, is_async=is_async)

    if function is not None:
        return _create_sync_atomic_wrapper(
            cast("Callable[..., R]", function),
            db_type=db_type,
            atomic_flag=atomic_flag,
        )
    return partial(sqlalchemy_atomic_decorator, db_type=db_type, is_async=is_async)

archipy.helpers.decorators.sqlalchemy_atomic.postgres_sqlalchemy_atomic_decorator

postgres_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for PostgreSQL atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def postgres_sqlalchemy_atomic_decorator(function: Callable[..., Any] | None = None) -> Callable[..., Any] | partial:
    """Decorator for PostgreSQL atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="postgres", function=function)

archipy.helpers.decorators.sqlalchemy_atomic.async_postgres_sqlalchemy_atomic_decorator

async_postgres_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for asynchronous PostgreSQL atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def async_postgres_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial:
    """Decorator for asynchronous PostgreSQL atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="postgres", is_async=True, function=function)

archipy.helpers.decorators.sqlalchemy_atomic.sqlite_sqlalchemy_atomic_decorator

sqlite_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for SQLite atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def sqlite_sqlalchemy_atomic_decorator(function: Callable[..., Any] | None = None) -> Callable[..., Any] | partial:
    """Decorator for SQLite atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="sqlite", function=function)

archipy.helpers.decorators.sqlalchemy_atomic.async_sqlite_sqlalchemy_atomic_decorator

async_sqlite_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for asynchronous SQLite atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def async_sqlite_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial:
    """Decorator for asynchronous SQLite atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="sqlite", is_async=True, function=function)

archipy.helpers.decorators.sqlalchemy_atomic.starrocks_sqlalchemy_atomic_decorator

starrocks_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for StarRocks atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def starrocks_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial:
    """Decorator for StarRocks atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="starrocks", function=function)

archipy.helpers.decorators.sqlalchemy_atomic.async_starrocks_sqlalchemy_atomic_decorator

async_starrocks_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for asynchronous StarRocks atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def async_starrocks_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial:
    """Decorator for asynchronous StarRocks atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="starrocks", is_async=True, function=function)

archipy.helpers.decorators.sqlalchemy_atomic.mysql_sqlalchemy_atomic_decorator

mysql_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for MySQL atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def mysql_sqlalchemy_atomic_decorator(function: Callable[..., Any] | None = None) -> Callable[..., Any] | partial:
    """Decorator for MySQL atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="mysql", function=function)

archipy.helpers.decorators.sqlalchemy_atomic.async_mysql_sqlalchemy_atomic_decorator

async_mysql_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial

Decorator for asynchronous MySQL atomic transactions.

Parameters:

Name	Type	Description	Default
function	Callable | None	The function to wrap. If None, returns a partial function.	None

Returns:

Type	Description
Callable[..., Any] | partial	Callable | partial: The wrapped function or a partial function for later use.

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

def async_mysql_sqlalchemy_atomic_decorator(
    function: Callable[..., Any] | None = None,
) -> Callable[..., Any] | partial:
    """Decorator for asynchronous MySQL atomic transactions.

    Args:
        function (Callable | None): The function to wrap. If None, returns a partial function.

    Returns:
        Callable | partial: The wrapped function or a partial function for later use.
    """
    return sqlalchemy_atomic_decorator(db_type="mysql", is_async=True, function=function)

options: show_root_toc_entry: false heading_level: 3

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.P module-attribute

P = ParamSpec('P')

archipy.helpers.decorators.deprecation_warnings.R module-attribute

R = TypeVar('R')

archipy.helpers.decorators.deprecation_warnings.T module-attribute

T = TypeVar('T', bound=type[Any])

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(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(message: str | None = None) -> Callable[[T], T]:
    """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 = cls.__init__

        def new_init(self: Any, *args: Any, **kwargs: Any) -> 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.P module-attribute

P = ParamSpec('P')

archipy.helpers.decorators.deprecation_exception.R module-attribute

R = TypeVar('R')

archipy.helpers.decorators.deprecation_exception.T module-attribute

T = TypeVar('T', bound=type[Any])

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(
    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(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:
        def new_init(_self: Any, *_args: Any, **_kwargs: Any) -> 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.P module-attribute

P = ParamSpec('P')

archipy.helpers.decorators.grpc_rate_limit.R module-attribute

R = TypeVar('R')

archipy.helpers.decorators.grpc_rate_limit.grpc_rate_limit_decorator

grpc_rate_limit_decorator(
    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.py

def grpc_rate_limit_decorator(
    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