Skip to content

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.

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 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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 
103
104
105
106
107
108
109
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 = 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

Time to live in seconds (default: 5 minutes). 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.
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def ttl_cache_decorator[**P, R](
    ttl_seconds: 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).
            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.

    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

    cache: TTLCache = TTLCache(maxsize=maxsize, ttl=ttl_seconds)

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

    return decorator  # type: ignore[return-value]

options: show_root_toc_entry: false heading_level: 3

Retry

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

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 = 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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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.

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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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.

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

logger = 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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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

Decorator that adds distributed tracing instrumentation to a function.

Tracing decorators for capturing transactions and spans in pure Python applications.

This module provides decorators to instrument code with APM tracing when not using gRPC or FastAPI frameworks. Supports both Sentry and Elastic APM based on configuration.

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

logger = getLogger(__name__)

archipy.helpers.decorators.tracing.capture_transaction 

capture_transaction(
    name: str | None = None,
    *,
    op: str = "function",
    description: str | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorator to capture a transaction for the decorated function.

This decorator creates a transaction span around the execution of the decorated function. It integrates with both Sentry and Elastic APM based on the application configuration.

Parameters:

Name Type Description Default
name str | None

Name of the transaction. If None, uses the function name.

 None
op str

Operation type/category for the transaction. Defaults to "function".

 'function'
description str | None

Optional description of the transaction.

 None

Returns:

Type Description
Callable[[F], Callable[..., Any]]

The decorated function with transaction tracing capabilities.
Example 
@capture_transaction(name="user_processing", op="business_logic")
def process_user_data(user_id: int) -> dict[str, Any]:
    # Your business logic here
    return {"user_id": user_id, "status": "processed"}


# Transaction will be automatically captured when function is called
result = process_user_data(123)
Source code in archipy/helpers/decorators/tracing.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def capture_transaction[F: Callable[..., Any]](
    name: str | None = None,
    *,
    op: str = "function",
    description: str | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorator to capture a transaction for the decorated function.

    This decorator creates a transaction span around the execution of the decorated function.
    It integrates with both Sentry and Elastic APM based on the application configuration.

    Args:
        name: Name of the transaction. If None, uses the function name.
        op: Operation type/category for the transaction. Defaults to "function".
        description: Optional description of the transaction.

    Returns:
        The decorated function with transaction tracing capabilities.

    Example:
        ```python
        @capture_transaction(name="user_processing", op="business_logic")
        def process_user_data(user_id: int) -> dict[str, Any]:
            # Your business logic here
            return {"user_id": user_id, "status": "processed"}


        # Transaction will be automatically captured when function is called
        result = process_user_data(123)
        ```
    """

    def decorator(func: F) -> Callable[..., Any]:
        transaction_name = name or func.__name__

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

            # Initialize and track with Sentry if enabled
            sentry_transaction = None
            if config.SENTRY.IS_ENABLED:
                try:
                    import sentry_sdk

                    # Initialize Sentry if not already done
                    current_hub = sentry_sdk.Hub.current
                    if not getattr(current_hub, "client", None):
                        sentry_sdk.init(
                            dsn=config.SENTRY.DSN,
                            debug=config.SENTRY.DEBUG,
                            release=config.SENTRY.RELEASE,
                            sample_rate=config.SENTRY.SAMPLE_RATE,
                            traces_sample_rate=config.SENTRY.TRACES_SAMPLE_RATE,
                            environment=getattr(config, "ENVIRONMENT", None),
                        )
                    sentry_transaction = sentry_sdk.start_transaction(
                        name=transaction_name,
                        op=op,
                        description=description or transaction_name,
                    )
                    sentry_transaction.__enter__()
                except ImportError:
                    logger.debug("sentry_sdk is not installed, skipping Sentry transaction capture.")
                except Exception:
                    logger.exception("Failed to initialize Sentry or start transaction")

            # Initialize and track with Elastic APM if enabled
            elastic_client: Any = None
            if config.ELASTIC_APM.IS_ENABLED:
                try:
                    import elasticapm

                    # Initialize Elastic APM client with config
                    elastic_client = elasticapm.get_client()
                    if not elastic_client:
                        elastic_client = elasticapm.Client(config.ELASTIC_APM.model_dump())
                    elastic_client.begin_transaction(transaction_type="function")
                except ImportError:
                    logger.debug("elasticapm is not installed, skipping Elastic APM transaction capture.")
                except Exception:
                    logger.exception("Failed to initialize Elastic APM or start transaction")
                    elastic_client = None

            try:
                # Execute the function
                result = func(*args, **kwargs)
            except Exception:
                # Mark transaction as failed and capture the exception
                if sentry_transaction:
                    sentry_transaction.set_status("internal_error")
                if elastic_client is not None:
                    elastic_client.end_transaction(name=transaction_name, result="error")

                # Re-raise the exception
                raise
            else:
                # Mark transaction as successful
                if sentry_transaction:
                    sentry_transaction.set_status("ok")
                if elastic_client is not None:
                    elastic_client.end_transaction(name=transaction_name, result="success")
                return result
            finally:
                # Clean up Sentry transaction
                if sentry_transaction:
                    try:
                        sentry_transaction.__exit__(None, None, None)
                    except Exception:
                        logger.exception("Error closing Sentry transaction")

        # @wraps preserves the function signature, making wrapper compatible with F
        wrapper.__wrapped__ = func
        return wrapper

    return decorator

archipy.helpers.decorators.tracing.capture_span 

capture_span(
    name: str | None = None,
    *,
    op: str = "function",
    description: str | None = None,
) -> Callable[[F], Callable[..., Any]]

Decorator to capture a span for the decorated function.

This decorator creates a span around the execution of the decorated function. Spans are child operations within a transaction and help provide detailed performance insights. Works with both Sentry and Elastic APM.

Parameters:

Name Type Description Default
name str | None

Name of the span. If None, uses the function name.

 None
op str

Operation type/category for the span. Defaults to "function".

 'function'
description str | None

Optional description of the span.

 None

Returns:

Type Description
Callable[[F], Callable[..., Any]]

The decorated function with span tracing capabilities.
Example 
@capture_transaction(name="user_processing")
def process_user_data(user_id: int) -> dict[str, Any]:
    user = get_user(user_id)
    processed_data = transform_data(user)
    save_result(processed_data)
    return processed_data


@capture_span(name="database_query", op="db")
def get_user(user_id: int) -> dict[str, Any]:
    # Database query logic here
    return {"id": user_id, "name": "John"}


@capture_span(name="data_transformation", op="processing")
def transform_data(user: dict[str, Any]) -> dict[str, Any]:
    # Data transformation logic
    return {"processed": True, **user}


@capture_span(name="save_operation", op="db")
def save_result(data: dict[str, Any]) -> None:
    # Save logic here
    pass
Source code in archipy/helpers/decorators/tracing.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def capture_span[F: Callable[..., Any]](
    name: str | None = None,
    *,
    op: str = "function",
    description: str | None = None,
) -> Callable[[F], Callable[..., Any]]:
    """Decorator to capture a span for the decorated function.

    This decorator creates a span around the execution of the decorated function.
    Spans are child operations within a transaction and help provide detailed
    performance insights. Works with both Sentry and Elastic APM.

    Args:
        name: Name of the span. If None, uses the function name.
        op: Operation type/category for the span. Defaults to "function".
        description: Optional description of the span.

    Returns:
        The decorated function with span tracing capabilities.

    Example:
        ```python
        @capture_transaction(name="user_processing")
        def process_user_data(user_id: int) -> dict[str, Any]:
            user = get_user(user_id)
            processed_data = transform_data(user)
            save_result(processed_data)
            return processed_data


        @capture_span(name="database_query", op="db")
        def get_user(user_id: int) -> dict[str, Any]:
            # Database query logic here
            return {"id": user_id, "name": "John"}


        @capture_span(name="data_transformation", op="processing")
        def transform_data(user: dict[str, Any]) -> dict[str, Any]:
            # Data transformation logic
            return {"processed": True, **user}


        @capture_span(name="save_operation", op="db")
        def save_result(data: dict[str, Any]) -> None:
            # Save logic here
            pass
        ```
    """

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

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

            # Track with Sentry if enabled
            sentry_span = None
            if config.SENTRY.IS_ENABLED:
                try:
                    import sentry_sdk

                    sentry_span = sentry_sdk.start_span(
                        op=op,
                        description=span_name,
                    )
                    sentry_span.__enter__()
                except ImportError:
                    logger.debug("sentry_sdk is not installed, skipping Sentry span capture.")

            # Track with Elastic APM if enabled
            elastic_client: Any = None
            elastic_span: Any = None
            if config.ELASTIC_APM.IS_ENABLED:
                try:
                    import elasticapm

                    elastic_client = elasticapm.get_client()
                    if elastic_client:
                        # begin_span is a valid method on elasticapm.Client
                        begin_span_method = getattr(elastic_client, "begin_span", None)
                        if begin_span_method is not None:
                            elastic_span = begin_span_method(
                                name=span_name,
                                span_type=op,
                            )
                except ImportError:
                    logger.debug("elasticapm is not installed, skipping Elastic APM span capture.")

            try:
                # Execute the function
                result = func(*args, **kwargs)
            except Exception as e:
                # Mark span as failed and capture the exception
                if sentry_span:
                    sentry_span.set_status("internal_error")

                # Add exception context to spans
                if sentry_span:
                    sentry_span.set_tag("error", True)
                    sentry_span.set_data("exception", str(e))

                if elastic_span and elastic_client:
                    elastic_client.capture_exception()

                # Re-raise the exception
                raise
            else:
                # Mark span as successful
                if sentry_span:
                    sentry_span.set_status("ok")
                return result
            finally:
                # Clean up spans
                if elastic_span and elastic_client:
                    try:
                        elastic_client.end_span()
                    except Exception:
                        logger.exception("Error closing Elastic APM span")

                if sentry_span:
                    try:
                        sentry_span.__exit__(None, None, None)
                    except Exception:
                        logger.exception("Error closing Sentry span")

        # @wraps preserves the function signature, making wrapper compatible with F
        wrapper.__wrapped__ = func
        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.

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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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).

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

logger = 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",
    },
}

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: bool = False,
    function: Callable[..., R] = ...,
) -> Callable[..., 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] | None = None,
) -> Callable[..., 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", or "starrocks").

 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] | 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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def sqlalchemy_atomic_decorator[R](
    db_type: str,
    is_async: bool = False,
    function: Callable[..., R] | None = None,
) -> Callable[..., 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", or "starrocks").
        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 ValueError(f"Invalid db_type: {db_type}. Must be one of {list(ATOMIC_BLOCK_CONFIGS.keys())}")

    atomic_flag = ATOMIC_BLOCK_CONFIGS[db_type]["flag"]

    # Dynamically import the registry class
    def get_registry() -> type[SessionManagerRegistry]:
        """Get the session manager registry for the specified database type.

        Returns:
            type[SessionManagerRegistry]: The session manager registry class.

        Raises:
            DatabaseConfigurationError: If the registry cannot be loaded.
        """
        try:
            import importlib

            module_path, class_name = ATOMIC_BLOCK_CONFIGS[db_type]["registry"].rsplit(".", 1)
            module = importlib.import_module(module_path)
            registry_class = getattr(module, class_name)
            if not isinstance(registry_class, type) or not issubclass(registry_class, SessionManagerRegistry):
                raise DatabaseConfigurationError(
                    database=db_type,
                    additional_data={"registry_path": ATOMIC_BLOCK_CONFIGS[db_type]["registry"]},
                )
        except (ImportError, AttributeError) as e:
            raise DatabaseConfigurationError(
                database=db_type,
                additional_data={"registry_path": ATOMIC_BLOCK_CONFIGS[db_type]["registry"]},
            ) from e
        else:
            return registry_class

    if is_async:

        def async_decorator(func: Callable[..., Awaitable[R]]) -> Callable[..., Awaitable[R]]:
            """Create an async transaction-aware wrapper for the given function.

            Args:
                func: The async function to wrap with transaction management.

            Returns:
                The wrapped async function that manages transactions.
            """

            @wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> R:
                """Async wrapper for managing database transactions."""
                registry = get_registry()
                session_manager: AsyncSessionManagerPort = registry.get_async_manager()
                session = session_manager.get_session()
                is_nested = session.info.get(atomic_flag, False)
                if not is_nested:
                    session.info[atomic_flag] = True

                try:
                    if session.in_transaction():
                        result = await func(*args, **kwargs)
                        if not is_nested:
                            await session.commit()
                        return result
                    else:
                        async with session.begin():
                            result = await func(*args, **kwargs)
                            return result
                except BaseException as exception:
                    await session.rollback()
                    func_name = getattr(func, "__name__", "unknown")
                    _handle_db_exception(exception, db_type, func_name)
                    # _handle_db_exception always raises, but add this for type checker
                    raise
                finally:
                    if not session.in_transaction():
                        await session.close()
                        await session_manager.remove_session()

            return async_wrapper

        if function is not None:
            return async_decorator(function)  # type: ignore[arg-type, return-value]
        return partial(sqlalchemy_atomic_decorator, db_type=db_type, is_async=is_async)

    else:

        def sync_decorator(func: Callable[..., R]) -> Callable[..., R]:
            """Create a sync transaction-aware wrapper for the given function.

            Args:
                func: The sync function to wrap with transaction management.

            Returns:
                The wrapped sync function that manages transactions.
            """

            @wraps(func)
            def sync_wrapper(*args: Any, **kwargs: Any) -> R:
                """Synchronous wrapper for managing database transactions."""
                registry = get_registry()
                session_manager: SessionManagerPort = registry.get_sync_manager()
                session = session_manager.get_session()
                is_nested = session.info.get(atomic_flag, False)
                if not is_nested:
                    session.info[atomic_flag] = True

                try:
                    if session.in_transaction():
                        result = func(*args, **kwargs)
                        if not is_nested:
                            session.commit()
                        return result
                    else:
                        with session.begin():
                            return func(*args, **kwargs)
                except BaseException as exception:
                    session.rollback()
                    func_name = getattr(func, "__name__", "unknown")
                    _handle_db_exception(exception, db_type, func_name)
                    # _handle_db_exception always raises, but add this for type checker
                    raise
                finally:
                    if not session.in_transaction():
                        session.close()
                        session_manager.remove_session()

            return sync_wrapper

        if function is not None:
            return sync_decorator(function)
        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 
312
313
314
315
316
317
318
319
320
321
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 
324
325
326
327
328
329
330
331
332
333
334
335
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 
338
339
340
341
342
343
344
345
346
347
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 
350
351
352
353
354
355
356
357
358
359
360
361
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 
364
365
366
367
368
369
370
371
372
373
374
375
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 
378
379
380
381
382
383
384
385
386
387
388
389
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)

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.

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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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)

        # @wraps preserves the function signature, making wrapper compatible with the original function
        wrapper.__wrapped__ = func
        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 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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.

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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def 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