Timeout

A Timeout policy bounds how long an async call may run. Use it to keep a slow dependency from blocking a request indefinitely.

Why

• Cap the worst-case latency of a downstream call.
• Reconfigure the deadline at runtime from a ConfigMap without redeploying.
• Compose with Retry, CircuitBreaker, Bulkhead, and Fallback.

Timeout wraps asyncio.timeout. When the deadline elapses, the inner block is cancelled and TimeoutError is raised.

Usage

from grelmicro.resilience import Timeout

db_timeout = Timeout("db", seconds=2.0)


async def fetch_rows(db) -> list[dict]:
    async with db_timeout:
        return await db.fetch_all("SELECT * FROM accounts")

The policy works as an async context manager and as a decorator on async functions. Sync functions are not supported (asyncio cannot cancel sync code).

from grelmicro.resilience import Timeout

db_timeout = Timeout("db", seconds=2.0)


@db_timeout
async def fetch_rows(db) -> list[dict]:
    return await db.fetch_all("SELECT * FROM accounts")

Configuration

Build the policy with keyword arguments. Set seconds to the deadline for the wrapped call.

from grelmicro.resilience import Timeout

db_timeout = Timeout("db", seconds=2.0)

Environment variables

Prefix: GREL_TIMEOUT_{NAME_UPPER}_. The default instance drops the name segment and reads GREL_TIMEOUT_*.

Env var	Field	Type	Default
GREL_TIMEOUT_{NAME_UPPER}_SECONDS	seconds	PositiveFloat	required

from grelmicro.resilience import Timeout

# Reads the deadline from environment variables.
#
# - GREL_TIMEOUT_DB_SECONDS=2.0
db_timeout = Timeout("db")

Advanced

For the from_config declarative path and pydantic-settings composition, see Declarative configuration.

Composition

The recommended outside-in order is Fallback → Retry → CircuitBreaker → Bulkhead → Timeout → call. Read more in Composing patterns.

import httpx

from grelmicro.resilience import CircuitBreaker, Retry, Timeout, fallback

breaker = CircuitBreaker("recs")
retrier = Retry.exponential("recs", when=httpx.HTTPError, attempts=3)
call_timeout = Timeout("recs", seconds=1.0)


@fallback(when=Exception, default=[])
@retrier
@breaker
@call_timeout
async def get_recommendations(
    client: httpx.AsyncClient, user_id: str
) -> list[dict]:
    response = await client.get(f"/recs/{user_id}")
    response.raise_for_status()
    return response.json()

A per-attempt Timeout placed under Retry shrinks the deadline of each retry while the retry budget stays the same.

Live reconfiguration

Timeout inherits Reconfigurable[TimeoutConfig]. Calling policy.reconfigure(new_config) swaps the deadline for future entries. Scopes already inside async with keep their original deadline. See Live reconfiguration.

Reference

See the API reference for every option.