Choosing a decorator¶

Find your problem, use the decorator in the second column.

Things fail¶

Problem	Use	Guide
A network call fails now and then	`@retry(on=ConnectionError, delay=..., backoff=2)`	Resilience
I want a default value instead of an exception	`@fallback(value)` or `@fallback(handler=fn)`	Resilience
I just want an error to go away (returns `None`)	`@suppress(KeyError)`	Resilience
A dependency is down and I keep hammering it	`@circuit_breaker(...)` or a shared `CircuitBreaker`	Resilience
A call can hang forever	`@timeout(seconds)` or `with timeout(seconds):`	Resilience
The server says `Retry-After`	`@retry(delay_from=retry_after_header)`	Resilience
Retry only part of a function	`for attempt in retrying(...)`	Resilience
A whole request needs one time budget	`with deadline(seconds):`	Resilience
A read is occasionally very slow	`@hedge(delay)` (idempotent calls only)	Batching and hedging
Apply retry, breaker and timeout in the right order	`policy(retry=..., breaker=..., timeout=...)`	Policies

Things are slow or repeated¶

Problem	Use	Guide
Same arguments, same answer, compute it once	`@memoize`	Caching
Cached values should go stale after N seconds	`@memoize(ttl=...)`	Caching
Expensive one-time setup	`@once`	Caching
100 concurrent requests for the same thing	`@singleflight`	Concurrency
An expensive attribute on an object	`@cached_property(ttl=...)`	Caching
Run a blocking function in the background	`@run_in_thread`	Concurrency

Too many things at once¶

Problem	Use	Guide
`gather` thousands of coroutines without flooding a server	`await gather_limit(*coros, limit=20)`	Async toolkit
Apply an async function to a list, a few at a time	`await map_async(fn, items, limit=20)`	Async toolkit
Take whichever of several calls answers first	`await race(a(), b())`	Async toolkit
Call async code from a script	`run_sync(coro)`	Async toolkit
Protect a dependency from overload; fail fast when full	`Bulkhead(n, max_waiting=...)` or `@limit_concurrency(n)`	Async toolkit
Many single lookups should become one bulk query	`@batch_calls(max_size=..., max_wait=...)`	Batching and hedging

Things happen too often¶

Problem	Use	Guide
Reject calls over N per second	`@rate_limit(n, period)`	Concurrency
Slow callers down instead of rejecting them	`@rate_limit(n, period, block=True)`	Concurrency
One limit per user or API key	`@rate_limit(..., key=...)`	Concurrency
Several functions share one API quota	`RateLimiter(n, period)`	Concurrency
Run at most once per interval, skip the rest	`@throttle(interval)`	Concurrency
Wait for a burst of calls to settle, then run once	`@debounce(wait)`	Concurrency

I need to see what is happening¶

Problem	Use	Guide
How long does this take?	`@timing`	Observability
Send durations to my metrics system	`@timing(callback=...)`	Observability
Compare how fast a function is over many runs	`benchmark(func, ...)`	Observability
How much memory does it allocate?	`@measure_memory`	Observability
Where does the time go?	`@profile`	Observability
Log the arguments (without leaking passwords)	`@log_args`	Observability
Log what a function returns	`@log_return`	Observability
Keep an audit trail of who called what	`@audit`	Observability
Log the call stack that led here	`@trace`	Observability
Count calls, errors and p95 latency per function	`@count_calls` then `func.stats`	Metrics and events
Feed retries, cache hits and breaker trips to Prometheus or OpenTelemetry	`subscribe(listener)`	Metrics and events
Put a request id on every log line	`with log_context(request_id=...):`	Observability

Bad input¶

Problem	Use	Guide
Check arguments with my own rules	`@validate_args(x=lambda x: x > 0)`	Validation
`isinstance` checks on a few arguments	`@check_type(x=int)`	Validation
Enforce my type annotations at runtime	`@enforce_types`	Validation

Half-finished work¶

Problem	Use	Guide
Undo changes to `self` if a method raises	`@revert_on_failure`	State safety
Register undo steps and run them if anything fails	`@transactional` / `Transaction`	State safety

Maintaining and testing code¶

Problem	Use	Guide
Tests that wait on retries and backoff	`with fast_mode():` from `python_wrap.testing`	Testing
Switch retries, rate limits or caches off in a test	`disable_retries()`, `disable_rate_limits()`, `disable_caching()`	Testing
Control time in a test	`FakeClock` / `fake_clock()`	Testing
Set the same logger, retry defaults or redaction everywhere	`configure(...)`	Configuration
Protect every method of a class	`@wrap_methods(...)`	Developer tools
Warn callers that an API is going away	`@deprecate(since=..., alternative=...)`	Developer tools
Return fixed data in tests or local runs	`@mock_data(...)`	Developer tools
Reuse a stack of decorators	`compose(a, b, c)`	Developer tools

Common recipes¶

Robust HTTP call

from python_wrap import circuit_breaker, retry, timeout

@circuit_breaker(failure_threshold=5, recovery_timeout=30)
@retry(max_attempts=3, delay=0.2, backoff=2, jitter=0.1, on=(ConnectionError, TimeoutError))
@timeout(5)
def call_api() -> dict:
    ...

Each attempt gets its own 5 second deadline, and the breaker counts a call as failed only once its retries are used up. See Stacking decorators.

Cached, de-duplicated lookup

from python_wrap import memoize, singleflight

@memoize(ttl=30)
@singleflight
def load_price(sku: str) -> float:
    ...

Public API with input checks

from python_wrap import enforce_types, log_args, validate_args

@log_args
@enforce_types
@validate_args(amount=(lambda a: a > 0, "amount must be positive"))
def charge(account: str, amount: float) -> None:
    ...

The same thing as one named policy

from python_wrap import CircuitBreaker, policy, retry, retry_after_header, timeout

api = policy(
    breaker=CircuitBreaker(failure_threshold=5, recovery_timeout=30),
    retry=retry(max_attempts=3, delay=0.2, backoff=2, on=ConnectionError,
                delay_from=retry_after_header),
    timeout=timeout(5),
    name="api",
)

@api
def call_api() -> dict:
    ...

The roles decide the order, so listing them differently changes nothing. See Policies.