baldur.decorators — Resilience Decorators

Single-stage decorators, @protected presets, and orthogonal gates. The composed primitive @protected / @aprotected lives at the top level; this module hosts the standalone retry stage (@retry), preset compositions (@dlq_protect) and orthogonal call-site gates (@idempotent, @rate_limit, @domain_tag).

Decorators

retry

retry(
    domain: str = "default",
    max_attempts: int | None = None,
    retryable_exceptions: (
        tuple[type[Exception], ...] | None
    ) = None,
    non_retryable_exceptions: (
        tuple[type[Exception], ...] | None
    ) = None,
    backoff: BackoffStrategy | None = None,
    retry_on_result: Callable[[Any], bool] | None = None,
    max_elapsed: float | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Unified retry decorator — dual-dispatches on sync vs async functions.

Replaces the previous split retry decorators (one for def, one for async def) with a single call-style-safe surface. Both branches derive their configuration from RetryPolicyConfig.from_settings(domain) with the passed overrides applied:

An async def is wrapped by :class:AsyncRetryPolicy.
A plain def is wrapped by the synchronous RetryPolicy.

On exhaustion, both branches raise MaxRetriesExceededError (carrying last_error); success returns the unwrapped value. A call refused by a shared 429 cooldown that outlasts the wait budget raises RateLimitDeferredError instead — the function was never called, and not_before says when it may be. functools.wraps preserves the wrapped signature, so framework dependency injection (e.g. FastAPI Depends) resolves against the original parameters.

Outbound 429 coordination applies to both branches, and only when domain is set. A function with a named domain shares a cooldown with every other caller on that domain, so a 429 backs the fleet off together instead of each worker retrying on its own ladder. Leaving domain unset opts out — the placeholder is shared by every unnamed caller, and one cooldown record cannot stand for unrelated downstreams. On an async def the cooldown wait is an asyncio.sleep: it costs the call its latency and never stalls the event loop.

Parameters:

Name	Type	Description	Default
`domain`	`str`	Configuration domain (also the retry / DLQ / metric key, and the outbound 429 coordination key — see above).	`'default'`
`max_attempts`	`int \| None`	Override the total attempt count (`None` uses settings).	`None`
`retryable_exceptions`	`tuple[type[Exception], ...] \| None`	Override the retryable exception tuple.	`None`
`non_retryable_exceptions`	`tuple[type[Exception], ...] \| None`	Override the non-retryable exception tuple.	`None`
`backoff`	`BackoffStrategy \| None`	Explicit BackoffStrategy (`None` derives one from settings).	`None`
`retry_on_result`	`Callable[[Any], bool] \| None`	Synchronous predicate that returns True for a soft-error result (200 + error payload, `None`, partial response) that should be retried. An `async def` predicate raises `TypeError`.	`None`
`max_elapsed`	`float \| None`	Cooperative wall-clock retry budget in seconds (`None` uses settings, where it also defaults to disabled).	`None`

Example::

@retry(domain="payment", max_attempts=3)
def call_external_api():
    return requests.post(...)

@retry(domain="payment", retry_on_result=lambda r: r.get("status") == "error")
def fetch_data():
    ...

dlq_protect

dlq_protect(
    name: str,
    *,
    fallback: (
        Callable[[], Any]
        | Callable[[], Awaitable[Any]]
        | None
    ) = None,
    timeout: float | None = _TIMEOUT_UNSET,
    context_from: (
        Callable[..., PolicyContext] | None | Literal[False]
    ) = _CONTEXT_FROM_UNSET
) -> Callable[[Callable[..., T]], Callable[..., T]]

@protected with zero-message-loss defaults pinned.

Equivalent to @protected(name, dlq=True, retry=True, circuit_breaker=True) but states the intent at the decoration site.

Parameters:

Name	Type	Description	Default
`name`	`str`	Identifier used for metrics/logging (passed to `protect()`).	required
`fallback`	`Callable[[], Any] \| Callable[[], Awaitable[Any]] \| None`	Optional fallback callable invoked when all branches fail.	`None`
`timeout`	`float \| None`	Per-call timeout in seconds. Omit to use `ProtectSettings` defaults; pass `None` to disable.	`_TIMEOUT_UNSET`
`context_from`	`Callable[..., PolicyContext] \| None \| Literal[False]`	Forwarded to `@protected`. Defaults (when omitted) to `None` → auto-extract: `order_id` / `user_id` and the full primitive payload from the wrapped function's arguments populate the DLQ entry without manual wiring. Pass `False` at privacy-sensitive callsites (e.g., `@dlq_protect("auth.verify_password", context_from=False)`) to skip capture. Pass a `Callable[..., PolicyContext]` for custom extraction logic.	`_CONTEXT_FROM_UNSET`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorator that auto-detects sync vs async and dispatches accordingly.

Usage::

@dlq_protect("orders.charge")
def charge(order_id: int) -> None:
    ...

@dlq_protect("orders.charge_async")
async def charge_async(order_id: int) -> None:
    ...

idempotent

idempotent(
    *,
    domain: (
        IdempotencyDomain | str
    ) = IdempotencyDomain.CUSTOM,
    key_args: list[str] | None = None,
    key_fn: (
        Callable[..., str | IdempotencyKey] | None
    ) = None,
    operation: str | None = None,
    ttl: timedelta | None = None,
    execution_ttl: timedelta | None = None
) -> Callable[[Callable[..., T]], Callable[..., T]]

Atomic do-not-run-twice decorator.

Parameters:

Name	Type	Description	Default
`domain`	`IdempotencyDomain \| str`	`IdempotencyDomain` (or its string value) used to namespace the cache key. Defaults to `IdempotencyDomain.CUSTOM`.	`CUSTOM`
`key_args`	`list[str] \| None`	List of parameter names to extract from the wrapped function's signature. The cache key combines the domain, the per-operation component (see `operation`), and the extracted values — so two different operations sharing a domain and key values do not consume each other's dedup verdicts. Only primitive types are allowed (int/str/bool/UUID/etc.) — see `_ALLOWED_KEY_ARG_TYPES`. Mutually exclusive with `key_fn`.	`None`
`key_fn`	`Callable[..., str \| IdempotencyKey] \| None`	`(args, *kwargs) -> str \| IdempotencyKey` callable that builds the key — the full-control escape hatch (no per-operation component is added). Mutually exclusive with `key_args`.	`None`
`operation`	`str \| None`	Per-operation key component for the `key_args` form. Defaults to the decorated function's `module.qualname`, so each function gets its own dedup key space with zero configuration. CAVEAT: the default identity is code-derived — renaming or moving the function resets that operation's dedup memory at the deploy. Set `operation=` explicitly for correctness-critical operations, and give two entry points the same label when they guard one logical operation (e.g. an HTTP handler and a worker both guarding the same charge). Must not contain `:` or `\|` (the key separators). Not accepted with `key_fn` (the key is verbatim).	`None`
`ttl`	`timedelta \| None`	Dedup memory window — how long a completed (or failed) operation is remembered, i.e. how long duplicates stay blocked after success. `None` uses the gate memory default (`BALDUR_IDEMPOTENCY_GATE_MEMORY_TTL_SECONDS`, 30 minutes unless tuned).	`None`
`execution_ttl`	`timedelta \| None`	In-flight execution window — how long a running claim is honored before a crashed attempt becomes retryable (and before a competing process may take the key over). `None` uses the gate execution default (30 minutes). Set `execution_ttl` >= the worst-case runtime of the wrapped function; a value below it risks a concurrent duplicate run via stale takeover.	`None`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorator that auto-detects sync vs async.

Raises (at decoration time): TypeError: If both key_args and key_fn are supplied; or if an annotated key_args parameter is non-primitive; or if neither is supplied; or if operation is empty / contains a separator / is combined with key_fn; or if ttl / execution_ttl is not a positive timedelta.

Raises (at call time): IdempotencyDuplicateError: On SKIP (already completed) or ABORT (in-flight collision). IdempotencyUnavailableError: When the cache is unavailable (e.g. Redis down) during the check and fail_open_on_cache_error is False (the default fail-closed posture); opting in treats the unverifiable check as CONTINUE instead. TypeError: If an unannotated key_args value resolves to a non-primitive at runtime; or key_fn returns an unsupported type.

Usage::

@idempotent(domain=IdempotencyDomain.EXTERNAL_SERVICE, key_args=["order_id"])
def charge(order_id: int) -> None:
    ...

@idempotent(key_fn=lambda payload: payload["request_id"])
async def handle(payload: dict) -> None:
    ...

@idempotent(
    domain=IdempotencyDomain.EXTERNAL_SERVICE,
    key_args=["order_id"],
    operation="billing.charge",   # shared label across entry points
    ttl=timedelta(hours=2),       # remember completions for 2 h
    execution_ttl=timedelta(minutes=5),  # worst-case runtime bound
)
def charge_from_worker(order_id: int) -> None:
    ...

rate_limit

rate_limit(
    *,
    max_requests: int,
    window_seconds: int = 60,
    key_fn: Callable[..., str] | None = None,
    raise_on_limit: bool = True
) -> Callable[[Callable[..., T]], Callable[..., T]]

Function-level sliding-window rate limiter.

Parameters:

Name	Type	Description	Default
`max_requests`	`int`	Maximum requests allowed within `window_seconds`. Required — missing raises `TypeError` at decoration time.	required
`window_seconds`	`int`	Sliding-window size in seconds. Defaults to 60.	`60`
`key_fn`	`Callable[..., str] \| None`	Optional `(args, *kwargs) -> str` callable that derives the per-call bucket key. When `None`, a single shared bucket keyed by `func.__qualname__` is used.	`None`
`raise_on_limit`	`bool`	When `True` (default), raise `RateLimitExceeded` on rejection. When `False`, return `None` on rejection (the wrapped function's value is returned on allowed paths).	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorator that auto-detects sync vs async.

Raises (at call time): RateLimitExceeded: When raise_on_limit=True and the limiter rejects the call.

Usage::

@rate_limit(max_requests=10, window_seconds=60)
def search(q: str) -> list[str]:
    ...

@rate_limit(max_requests=5, window_seconds=1, key_fn=lambda user_id: f"u:{user_id}")
async def fetch(user_id: int) -> dict:
    ...

domain_tag

domain_tag(domain: str) -> Callable[[F], F]

Domain-tagging decorator.

Sets the domain context while the wrapped function executes and restores the previous context on exit (both sync and async callables are supported).

Parameters:

Name	Type	Description	Default
`domain`	`str`	Domain name to apply for the duration of the call.	required

Returns:

Type	Description
`Callable[[F], F]`	Decorator that wraps the target callable.

Usage

@domain_tag("payment") def process_payment(): # Errors raised here are tagged with the "payment" domain. ...

@domain_tag("order") async def create_order(): # Async functions are supported. ...

Context manager

DomainContext

DomainContext(domain: str)

Domain context manager.

Sets the domain for the enclosed code block via with. On block exit, the previous domain context is restored automatically.

Usage

with DomainContext("payment"): # The domain is "payment" inside this block. process_payment()

The previous domain is restored on exit.

Attributes:

Name	Type	Description
`domain`		Domain name to apply for the block.

Initialize a DomainContext.

Parameters:

Name	Type	Description	Default
`domain`	`str`	Domain name to apply.	required

Helpers

get_current_domain

get_current_domain() -> str | None

Return the current domain from the active context.

Returns:

Type	Description
`str \| None`	Current domain name, or None if no context is set.

Usage

@domain_tag("payment") def process_payment(): domain = get_current_domain() print(f"Current domain: {domain}") # "payment"

clear_domain_context

clear_domain_context() -> None

Clear the current domain context.

Primarily for test cleanup. In application code, DomainContext and @domain_tag clean up automatically on scope exit.

Exceptions raised by the decorators above

IdempotencyDuplicateError

IdempotencyDuplicateError(
    message: str = "",
    *,
    key: str = "",
    domain: str = "",
    decision: str = ""
)

Bases: BaldurError

Raised by @idempotent on a detected duplicate or in-flight collision.

Inherits BaldurError directly (correctness contract, not a resilience stage). Non-retryable by default — outer @dlq_protect layers should treat this as a terminal signal.

RateLimitExceeded

RateLimitExceeded(
    message: str = "",
    *,
    key: str = "",
    limit: int = 0,
    window_seconds: int = 0,
    reset_at: int = 0
)

Bases: ResilienceError

Raised by @rate_limit when a call is rejected by the limiter.

Function-level rejection signal — distinct from RateLimitStorageError (storage-backend failure).

DomainValidationError

DomainValidationError(
    message: str = "",
    *,
    original_domain: str = "",
    reason: Any = None
)

Bases: BaldurError

Raised when a domain input string fails validation.

Carries the original (pre-normalization) input and a typed reject reason for downstream logging / metric labelling.

Modeled on RecoveryAdapterError: raised at validation sites that have a loud failure mode (decoration-time, where a CI/dev surface can recover via test or rename). Runtime APIs catch this and fall back to FALLBACK_DOMAIN instead of propagating.