Skip to content

Operations

Circuit breakers, maintenance mode and feature flags, switched at runtime. In fastapi_alcazaba.operations.

In the guide: Circuit breakers, Maintenance mode, Feature flags. In an app: the example.

fastapi_alcazaba.operations.FastAPIBreaker 

FastAPIBreaker(name: str, *, failure_threshold: int = 5, reset_timeout: float = 30.0, timeout: float | None = None, failures: tuple[type[BaseException], ...] = (Exception,), exclude: tuple[type[BaseException], ...] = (), metrics: Metrics | None = None)

A circuit breaker named name (for logs and metrics), as a decorator of async or sync functions, or around a block with attempt():

@breaker
async def fetch(...): ...

async with breaker.attempt():
    ...

Parameters:

Name Type Description Default
failure_threshold int

consecutive failures that open it.

 5
reset_timeout float

seconds it stays open before a trial call may go through.

 30.0
timeout float | None

seconds an async call may take; past them it's cancelled, counts as a failure and raises TimeoutError. None for no limit. Sync code can't be interrupted: with a timeout, the breaker refuses to guard it.

 None
failures tuple[type[BaseException], ...]

the exceptions that count as failures of the dependency.

 (Exception,)
exclude tuple[type[BaseException], ...]

exceptions that mean the dependency answered (a "not found" from its SDK): they count as successes, and go on to the caller like any other.

 ()
metrics Metrics | None

where to count calls and transitions; FastAPICache's by default.

 None

state property 

state: str

"closed", "open" or "half-open" (a trial call may go through).

retry_after property 

retry_after: float

Seconds until a trial call may go through; 0 unless it's open.

failure_threshold property 

failure_threshold: int

Consecutive failures that open it.

reset_timeout property 

reset_timeout: float

Seconds it stays open before a trial call may go through.

reset 

reset() -> None

Close it now, and forget the failures counted: when you know the dependency is back (an admin says so), or between tests. Only this process's breaker: each process has its own.

attempt 

attempt() -> _Attempt

A context manager for one call: async with (with the timeout) or with.

fastapi_alcazaba.operations.BreakerOpen 

BreakerOpen(name: str, retry_after: float)

Bases: HTTPException

The breaker is open: the call wasn't made. FastAPI answers it with a 503 and Retry-After; the response doesn't name the dependency.

fastapi_alcazaba.operations.FastAPIMaintenance 

FastAPIMaintenance(cache: Cache | None = None, *, refresh: float = 1.0, namespace: str = 'maintenance')

The maintenance switch, in the cache's backend.

Parameters:

Name Type Description Default
cache Cache | None

the Cache whose backend keeps the state; FastAPICache's by default, resolved at each call.

 None
refresh float

seconds each process keeps its copy of the state before asking again.

 1.0
namespace str

part of the key, after the cache's prefix (see Cache.state_key).

 'maintenance'

forget 

forget() -> None

Drop this process's copy of the state: the next request reads it from the backend.

enable async 

enable(message: str | None = None, retry_after: int | None = None) -> MaintenanceStatus

Turn maintenance on in every process (within refresh seconds), with the message (the detail of the 503s) and retry_after seconds to send. Raises BackendUnavailable when the backend fails: nothing changes then.

disable async 

disable() -> None

Turn maintenance off in every process (within refresh seconds). Raises BackendUnavailable when the backend fails.

status async 

status() -> MaintenanceStatus

The state in the backend now; this process's copy is updated with it. Raises BackendUnavailable when the backend fails.

current async 

current() -> MaintenanceStatus

This process's copy of the state, refreshed when it's older than refresh seconds; what the middleware checks on every request. Never raises: when the backend fails, the last state read (off if none).

fastapi_alcazaba.operations.MaintenanceMiddleware 

MaintenanceMiddleware(app: ASGIApp, *, maintenance: FastAPIMaintenance, allow_paths: Iterable[str] = ('/health',), allow: Allow | None = None)

While maintenance is on, answers every HTTP request with a 503 (its message as detail, and Retry-After when it has one) and closes WebSockets with 1013 (try again later), except:

app.add_middleware(
    MaintenanceMiddleware,
    maintenance=maintenance,
    allow_paths=("/health", "/admin"),
    allow=lambda request: client_ip(request) in OFFICE,
)

Parameters:

Name Type Description Default
maintenance FastAPIMaintenance

the switch.

 required
allow_paths Iterable[str]

path prefixes that are always served (/health covers /health/live), matched like the app's routes: without the root_path a proxy or a mount adds in front.

 ('/health',)
allow Allow | None

decides whether other requests go through during maintenance.

 None

allowed_path 

allowed_path(path: str) -> bool

Whether path is always served.

fastapi_alcazaba.operations.FastAPIFeatureFlags 

FastAPIFeatureFlags(cache: Cache | None = None, refresh: float = 5.0, *, namespace: str = 'flags')

Feature flags shared by every process through the cache's backend.

Parameters:

Name Type Description Default
cache Cache | None

the Cache whose backend keeps the flags; FastAPICache's by default, resolved at each call.

 None
refresh float

seconds each process keeps a definition before asking the backend again.

 5.0
namespace str

part of the keys, after the cache's prefix (see Cache.state_key).

 'flags'

forget 

forget() -> None

Drop this process's copies of the definitions: the next evaluations read them from the backend.

set async 

set(name: str, *, enabled: bool, percentage: float | None = None, allow: Iterable[str] = (), deny: Iterable[str] = ()) -> FeatureFlag

Define the flag name (replacing its definition) for every process, within refresh seconds. percentage (0 to 100) of the subjects get it, those in allow always do and those in deny never. Without a percentage everybody does, or only allow when given. Raises BackendUnavailable when the backend fails.

delete async 

delete(name: str) -> None

Remove the flag name: enabled gives the default again. Raises BackendUnavailable when the backend fails.

get async 

get(name: str) -> FeatureFlag | None

The definition of name in the backend now, None if it has none; this process's copy is updated with it. Raises BackendUnavailable when the backend fails.

enabled async 

enabled(name: str, subject: str | None = None, default: bool = False) -> bool

Whether the flag name is on for subject; default when it isn't defined, or the backend fails and this process never read it.

require 

require(name: str, subject: Subject | None = None, *, default: bool = False) -> Callable[[HTTPConnection], Awaitable[None]]

A FastAPI dependency that answers 404 (FastAPI's own "Not Found", so the flag isn't revealed) when the flag name is off for the request's subject, and closes WebSockets with 1008. For subjects that need other dependencies (the current user), call enabled from your own dependency instead.