scaffold-archipy-decorator

Scaffold ArchiPy Decorator

Scope

Only helpers/decorators/. Do not create utils or interceptors here.

Before writing files

1. Inspect existing decorators, call sites, and the installed ArchiPy version for a matching decorator.
2. Infer sync/async style and project naming from the target call site.
3. Ask only for unresolved behavior. Prefer an ArchiPy decorator whenever it fits.
4. Preserve existing decorator modules; do not overwrite.

Prefer ArchiPy

Examples:

• from archipy.helpers.decorators.cache import ttl_cache_decorator
• from archipy.helpers.decorators.sqlalchemy_atomic import postgres_sqlalchemy_atomic_decorator
• from archipy.helpers.decorators.sqlalchemy_atomic import async_postgres_sqlalchemy_atomic_decorator
• trace_span / trace_root (+ async twins) from archipy.helpers.decorators.tracing
• measure_duration / count_calls (+ async twins) from archipy.helpers.decorators.metrics
• timeout_decorator, retry_decorator, singleton_decorator, timing_decorator, and grpc_rate_limit_decorator under archipy.helpers.decorators

Show correct usage on a sample function; do not reimplement. UoW decorators belong on logics, not services/repositories.

Custom decorator

Create helpers/decorators/<name>.py:

from __future__ import annotations

import functools
import logging
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar

logger = logging.getLogger(__name__)

P = ParamSpec("P")
R = TypeVar("R")


def timed(func: Callable[P, R]) -> Callable[P, R]:
    """Log wall-clock duration of a sync call.

    Args:
        func: Callable to wrap.

    Returns:
        Wrapped callable that logs elapsed milliseconds.

    Example:
        @timed
        def build_report(order_id: str) -> str:
            ...
    """

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        started = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            elapsed_ms = (time.perf_counter() - started) * 1000
            logger.debug("%s took %.2f ms", func.__qualname__, elapsed_ms)

    return wrapper