Fallback
A Fallback policy returns a safe value when a call fails. Use it to degrade gracefully: "if the live recommendation service fails, return cached recommendations or an empty list".
Why
- Keep the user-facing path working when a dependency is down.
- Replace a hard error with a quiet default, without changing every call site.
- Compose with Retry and Circuit Breaker so the fallback only fires after retries are exhausted or the breaker is open.
Usage
import httpx
from grelmicro.resilience import fallback
@fallback(when=httpx.HTTPError, default=[])
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()
The decorator works on async and sync functions. It auto-detects which kind it wraps.
Use factory= instead of default= when the fallback value should be computed from the exception or read from a cache:
import httpx
from grelmicro.resilience import fallback
def _cached_recommendations(exc: BaseException) -> list[dict]:
# Read from a local cache shared by the process.
return []
@fallback(when=httpx.HTTPError, factory=_cached_recommendations)
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()
default= and factory= are mutually exclusive. Exactly one must be set.
For inline fallback that spans multiple statements, use the block form:
import httpx
from grelmicro.resilience import falling_back
async def get_recommendations(
client: httpx.AsyncClient, user_id: str
) -> list[dict]:
async with falling_back(when=httpx.HTTPError, default=[]) as result:
response = await client.get(f"/recs/{user_id}")
response.raise_for_status()
result.set(response.json())
return result.value
The block yields a FallbackResult. Call result.set(value) on success. If a matching exception is raised inside the block, it is suppressed and result.value is the configured default or the factory output.
Filtering exceptions
when= accepts the same shorthand as Retry's when=: an exception class, a tuple of classes, a predicate, or a Match instance.
from grelmicro.resilience import Fallback, Match
Fallback("api", when=httpx.HTTPError, default=None)
Fallback("api", when=(httpx.HTTPError, OSError), default=None)
Fallback("api", when=lambda e: e.status >= 500, default=None)
Fallback("api", when=Match.exception(httpx.HTTPError), default=None)
See Filtering outcomes with Match for the full DSL. Only exception-shaped matchers make sense for Fallback: result matching does not apply.
What never falls back
asyncio.CancelledError, KeyboardInterrupt, and SystemExit are BaseException subclasses outside Exception. They always propagate, regardless of when=. This is required for correct asyncio shutdown.
Configuration
Build the policy with keyword arguments. Set when to choose which exceptions fall back, and default or factory for the safe value.
import httpx
from grelmicro.resilience import Fallback
policy = Fallback("recs", when=httpx.HTTPError, default=[])
@policy
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()
Environment variables
Prefix: GREL_FALLBACK_{NAME_UPPER}_. The default instance drops the name segment and reads GREL_FALLBACK_*.
| Env var | Field | Type | Default |
|---|---|---|---|
GREL_FALLBACK_{NAME_UPPER}_WHEN |
when |
CSV or JSON list of FQN strings (e.g. httpx.HTTPError). Coerced to Match.exception(...). Predicate forms cannot come from env. |
required |
GREL_FALLBACK_{NAME_UPPER}_DEFAULT |
default |
JSON value ([], null, 42, ...). Strings that fail to parse are kept verbatim (hello stays "hello"). Mutually exclusive with factory=. |
mutually exclusive with factory |
The JSON parse runs only on env values. A default="[1,2,3]" passed in code stays a string. The factory callable cannot come from env. Use default for env-driven configs, or build the FallbackConfig in code.
from grelmicro.resilience import Fallback
# Reads the config from environment variables. `factory` cannot be set
# from env (it is a callable), use `default` or build the config from code.
#
# - GREL_FALLBACK_RECS_WHEN=httpx.HTTPError
# - GREL_FALLBACK_RECS_DEFAULT=[]
policy = Fallback("recs")
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, fallback
breaker = CircuitBreaker("recs")
retrier = Retry.exponential("recs", when=httpx.HTTPError, attempts=3)
@fallback(when=Exception, default=[])
@retrier
@breaker
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()
Live reconfiguration
Fallback inherits Reconfigurable[FallbackConfig]. Calling policy.reconfigure(new_config) swaps the snapshot for future calls. See Live reconfiguration.
Reference
See the API reference for every option.