Decorator Tutorials

This page demonstrates how to use ArchiPy's decorators for common cross-cutting concerns.

Retry Decorator

The retry decorator automatically retries a function when it encounters specific exceptions. It works for both regular and async def functions (async functions wait with asyncio.sleep, so the event loop is never blocked).

max_retries is the total number of attempts, including the first call, and must be at least 1. Use backoff (e.g. 2.0 doubles the delay after every failure) and jitter (e.g. 0.1 adds up to 10% random delay) to spread retries out. ignore takes precedence over retry_on.

import logging
import random
from typing import Any

from archipy.helpers.decorators import retry
from archipy.models.errors import ResourceExhaustedError

# Configure logging
logger = logging.getLogger(__name__)


# Retry a function that might fail temporarily
@retry(
    max_retries=3,
    delay=1,
    backoff=2.0,
    jitter=0.1,
    retry_on=(ConnectionError, TimeoutError),
    ignore=(ValueError,),
    resource_type="API",
)
def unreliable_api_call(item_id: int) -> dict[str, Any]:
    """Make an API call that might fail temporarily.

    Args:
        item_id: The ID of the item to fetch

    Returns:
        API response data

    Raises:
        ResourceExhaustedError: If retries are exhausted
        ValueError: If input validation fails (not retried)
    """
    if item_id < 0:
        # This error won't be retried because it's in the ignored list
        raise ValueError("Item ID must be positive")

    # Simulate random failures
    if random.random() < 0.7:
        # This will be retried because ConnectionError is in retry_on
        raise ConnectionError("Temporary network issue")

    # Success case
    return {"id": item_id, "name": f"Item {item_id}"}


try:
    # This call might succeed after retries
    result = unreliable_api_call(42)
except ResourceExhaustedError as e:
    # This happens when all retries fail
    logger.error(f"All retry attempts failed: {e}")
    raise
except ValueError as e:
    # This happens for input validation failures (not retried)
    logger.error(f"Validation error: {e}")
    raise
else:
    logger.info(f"Request succeeded: {result}")

Timeout Decorator

The timeout decorator ensures a function doesn't run longer than a specified duration. The limit may be a float (e.g. 0.5). Synchronous functions use SIGALRM, so they work on Unix and only from the main thread (calling one from another thread raises FailedPreconditionError instead of running without a limit); the previous signal handler is restored after the call. async def functions are bounded with asyncio.timeout and work in any thread.

import logging
import time

from archipy.helpers.decorators import timeout
from archipy.models.errors import DeadlineExceededError

# Configure logging
logger = logging.getLogger(__name__)


# Set a timeout for a potentially long-running function
@timeout(3)  # 3 seconds timeout
def slow_operation(duration: float) -> str:
    """A function that might take too long.

    Args:
        duration: How long to run in seconds

    Returns:
        Completion message

    Raises:
        DeadlineExceededError: If function takes longer than the timeout
    """
    time.sleep(duration)  # Simulate work
    return "Operation completed"


try:
    # This will succeed because it completes within the timeout
    result = slow_operation(2)
except DeadlineExceededError as e:
    logger.error(f"Operation timed out: {e}")
    raise
else:
    logger.info(result)  # "Operation completed"

try:
    # This will raise a DeadlineExceededError because it exceeds the timeout
    result = slow_operation(5)
except DeadlineExceededError as e:
    logger.error(f"Operation timed out: {e}")
    # Expected to timeout
else:
    logger.info("This won't be reached")

Timing Decorator

The timing decorator measures and logs the execution time of functions.

import logging
import time

from archipy.helpers.decorators import timing

# Configure logging
logger = logging.getLogger(__name__)


# Measure and log how long a function takes to execute
@timing
def process_data(items: list[int]) -> int:
    """Process a list of items with time measurement.

    Args:
        items: List of items to process

    Returns:
        Sum of processed items
    """
    time.sleep(0.1)  # Simulate processing time
    return sum(items)


# This will log the execution time before returning
result = process_data(list(range(100)))
logger.info(f"Result: {result}")  # Output: Result: 4950
# The decorator will log something like:
# INFO - Function 'process_data' executed in 0.103 seconds

Cache Decorator

The TTL cache decorator caches function results with automatic expiration.

import logging
import time

from archipy.helpers.decorators import ttl_cache

# Configure logging
logger = logging.getLogger(__name__)


# Cache the results of an expensive function
@ttl_cache(ttl_seconds=60, maxsize=100)
def fetch_user_data(user_id: int) -> dict[str, str | int]:
    """Fetch user data from a slow source with caching.

    Args:
        user_id: User ID to fetch

    Returns:
        User data dictionary
    """
    logger.info(f"Fetching data for user {user_id}...")
    time.sleep(1)  # Simulate slow API call
    return {
        "id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com"
    }


# First call - will execute the function and cache the result
start = time.time()
data1 = fetch_user_data(123)
logger.info(f"First call took {time.time() - start:.3f} seconds")

# Second call with same arguments - will use the cached result
start = time.time()
data2 = fetch_user_data(123)
logger.info(f"Second call took {time.time() - start:.3f} seconds")

# Different arguments - will execute the function
start = time.time()
data3 = fetch_user_data(456)
logger.info(f"Different user call took {time.time() - start:.3f} seconds")

# Clear the cache if needed
fetch_user_data.clear_cache()

SQLAlchemy Transaction Decorators

These decorators automatically manage database transactions.

Moved to the adapters layer

They now live in archipy.adapters.base.sqlalchemy.atomic. Importing them from archipy.helpers.decorators still works and emits a DeprecationWarning. The old import path will be removed in 6.0.0.

import logging
from uuid import UUID

from archipy.adapters.base.sqlalchemy.atomic import postgres_sqlalchemy_atomic
from archipy.models.errors import DatabaseQueryError

# Configure logging
logger = logging.getLogger(__name__)


@postgres_sqlalchemy_atomic
def create_user(username: str, email: str) -> "User":  # type: ignore[name-defined]
    """Create a user in a database transaction.

    All database operations are wrapped in a transaction that
    will be automatically committed on success or rolled back on error.

    Args:
        username: User's username
        email: User's email address

    Returns:
        The created user object

    Raises:
        DatabaseQueryError: If the database operation fails
    """
    user = User(username=username, email=email)  # type: ignore[name-defined]
    session = adapter.get_session()  # type: ignore[name-defined]
    session.add(user)
    logger.info(f"User created: {username}")
    return user


@postgres_sqlalchemy_atomic
async def update_user_email(user_id: UUID, new_email: str) -> "User | None":  # type: ignore[name-defined]
    """Update a user's email in an async transaction.

    Args:
        user_id: UUID of the user
        new_email: New email address

    Returns:
        Updated user or None if not found

    Raises:
        DatabaseQueryError: If the database operation fails
    """
    session = adapter.get_session()  # type: ignore[name-defined]
    user = await session.get(User, user_id)  # type: ignore[name-defined]
    if not user:
        logger.warning(f"User not found: {user_id}")
        return None

    user.email = new_email
    logger.info(f"Updated email for user: {user_id}")
    return user

Several databases

Databases of different types (for example PostgreSQL and MySQL) work out of the box: each type has its own registry, config section and decorator. An atomic block on one never joins a transaction on the other, and ArchiPy does not provide a distributed transaction, so a failure after the first commit does not undo it.

For a second database of the same type, give it its own config and its own registry, and pass the registry to the decorator:

from archipy.adapters.base.sqlalchemy.atomic import postgres_sqlalchemy_atomic
from archipy.adapters.postgres.sqlalchemy.adapters import PostgresSQLAlchemyAdapter
from archipy.adapters.postgres.sqlalchemy.session_manager_registry import PostgresSessionManagerRegistry
from archipy.adapters.postgres.sqlalchemy.session_managers import PostgresSQlAlchemySessionManager
from archipy.configs.config_template import PostgresSQLAlchemyConfig

analytics_config = PostgresSQLAlchemyConfig(HOST="analytics.internal", DATABASE="analytics", USERNAME="app", PASSWORD="...")


class AnalyticsRegistry(PostgresSessionManagerRegistry):
    """Session manager registry of the analytics database."""


AnalyticsRegistry.set_sync_manager(PostgresSQlAlchemySessionManager(analytics_config))
analytics = PostgresSQLAlchemyAdapter(analytics_config)  # shares the registry's session manager


@postgres_sqlalchemy_atomic(registry=AnalyticsRegistry)
def save_event(event: EventEntity) -> None:
    analytics.create(event)

A session manager is created once per distinct config, so the adapter and the registry that were built from the same config use the same engine and session. The default registry keeps reading the global config, so code that passes no registry is unchanged. sqlalchemy_atomic(db_type, registry=...) takes the same argument.

gRPC Rate Limit Decorator

grpc_rate_limit declares per-RPC rate limits on gRPC servicer methods. It does not enforce limits by itself — enable GRPC_RATE_LIMIT.IS_ENABLED (or register the interceptor manually) so decorated methods are checked before the handler runs.

from archipy.helpers.decorators import grpc_rate_limit
from archipy.helpers.utils.app_utils import AppUtils
from archipy.configs.base_config import BaseConfig

config = BaseConfig.global_config()

Set GRPC_RATE_LIMIT__IS_ENABLED=true so AppUtils.create_grpc_app / create_async_grpc_app register the matching interceptor automatically.

Sync servicer — def handlers + AppUtils.create_grpc_app:

class MySyncServiceServicer(pb2_grpc.MyServiceServicer):
    @grpc_rate_limit(calls_count=100, minutes=1)
    def Cheap(self, request, context):
        return pb2.CheapResponse()

    @grpc_rate_limit(calls_count=10, seconds=1)
    @grpc_rate_limit(calls_count=1000, days=1)
    def Expensive(self, request, context):
        return pb2.ExpensiveResponse()


sync_server = AppUtils.create_grpc_app(config)

Async servicer — async def handlers + AppUtils.create_async_grpc_app:

class MyAsyncServiceServicer(pb2_grpc.MyServiceServicer):
    @grpc_rate_limit(calls_count=100, minutes=1)
    async def Cheap(self, request, context):
        return pb2.CheapResponse()

    @grpc_rate_limit(calls_count=10, seconds=1)
    @grpc_rate_limit(calls_count=1000, days=1)
    async def Expensive(self, request, context):
        return pb2.ExpensiveResponse()


async_server = AppUtils.create_async_grpc_app(config)

Note: Use sync factories/interceptors with def handlers and async with async def. Manual customized_interceptors registration remains available when IS_ENABLED is false.

Parameters

Parameter	Default	Description
calls_count	1	Max requests allowed in the window
milliseconds	0	Window component (ms)
seconds	0	Window component (s)
minutes	0	Window component (min)
hours	0	Window component (h)
days	0	Window component (days)

Time units are summed into one window. At least one time component must be greater than zero.

Stacking

Apply multiple decorators on one method for burst + sustained tiers. The interceptor enforces windows in declaration order (inner decorator first) and short-circuits on the first breach.

Undecorated methods are never rate-limited.

Note: Redis 8.8+ required (INCREX). Install archipy[redis,grpc].

See Interceptor Tutorials — Rate Limiting for identity resolution, GRPC_RATE_LIMIT config, and fail-closed behavior.

See Also

• API Reference - Decorators - Full decorators API documentation
• API Reference - gRPC Rate Limit Interceptor - Interceptor API
• Helper Tutorials - Overview of all helper tutorials

Deprecated names

Decorators dropped their _decorator suffix and now detect async def functions themselves, so the separate async_* variants are no longer needed. The old names still work and emit a DeprecationWarning that names the replacement. They will be removed in 6.0.0.

Deprecated	Use instead
retry_decorator	retry
timeout_decorator	timeout
timing_decorator	timing
singleton_decorator	singleton
ttl_cache_decorator	ttl_cache
grpc_rate_limit_decorator	grpc_rate_limit
sqlalchemy_atomic_decorator(db_type, is_async=...)	sqlalchemy_atomic(db_type)
postgres_/mysql_/sqlite_/starrocks_sqlalchemy_atomic_decorator	postgres_/mysql_/sqlite_/starrocks_sqlalchemy_atomic
async_postgres_/async_mysql_/async_sqlite_/async_starrocks_..._decorator	postgres_/mysql_/sqlite_/starrocks_sqlalchemy_atomic
async_measure_duration, async_count_calls	measure_duration, count_calls
async_trace_span, async_trace_root	trace_span, trace_root

Import everything from archipy.helpers.decorators. The old submodule paths (for example archipy.helpers.decorators.retry) keep exporting the deprecated names until 6.0.0, which removes them with the other deprecated names above.