Skip to content

Getting started

Install

uv add interlock-cb

pip install interlock-cb

poetry add interlock-cb

The core is pure standard library. External integrations are optional extras — add the ones you need (same names with pip install / poetry add):

uv add 'interlock-cb[fastapi]'  # CircuitOpenError -> 503 + Retry-After
uv add 'interlock-cb[litestar]' # same for Litestar
uv add 'interlock-cb[httpx2]'   # per-host httpx2 transport
uv add 'interlock-cb[httpx]'    # per-host httpx transport
uv add 'interlock-cb[aiohttp]'  # per-host aiohttp client middleware
uv add 'interlock-cb[requests]' # per-host requests session adapter
uv add 'interlock-cb[tenacity]' # retry x breaker composition helpers
uv add 'interlock-cb[redis]'    # shared distributed state
uv add 'interlock-cb[otel]'     # OpenTelemetry metrics listener

Create a breaker

A breaker is named and configured once, then reused:

from interlock import CircuitBreaker, Config

breaker = CircuitBreaker(
    name='payments',
    config=Config(failure_rate_threshold=0.5, minimum_number_of_calls=20),
)

The defaults follow resilience4j: trip at a 50% failure rate over at least 10 calls, stay open for 60s, then admit up to 10 probe calls (one at a time) and decide from their outcomes. See Configuration for every option.

For an existing production dependency, start with initial_state=State.METRICS_ONLY: calls are always admitted while their outcomes populate the window. Tune thresholds from real traffic, then deploy a new breaker in the default CLOSED state. See Safe production rollout.

Three ways to protect work

All three run over the same call() primitive.

Decorator

@breaker
def charge(amount: int) -> str:
    return gateway.charge(amount)

The decorator preserves the wrapped signature and its sync/async nature — type checkers still see charge as (int) -> str.

breaker.call

result = breaker.call(gateway.charge, 100)

call inspects the callable to pick the sync or the async path. Where that is already known — a synchronous client, an awaited handler — call_sync and call_async skip the inspection:

result = breaker.call_sync(gateway.charge, 100)
result = await breaker.call_async(client.get, url)

call_sync never awaits, so passing a coroutine function to it records the coroutine's creation rather than its outcome. Reach for these on a hot path that calls the same shape every time; call is the right default everywhere else.

Context manager

with breaker:
    gateway.charge(100)

Contract difference

The decorator and call see a callable, so result-based classification and slow-call detection both apply. The context manager sees only the block — its exception and duration — so classification by return value is not available there. Need result-based classification? Use the decorator or call.

Async

The same instance handles async. The decorator and call detect a coroutine function; the instance is also an async context manager:

@breaker
async def fetch(url: str) -> bytes:
    return await client.get(url)


result = await breaker.call(client.get, url)

async with breaker:
    await client.get(url)

Handle rejections

When the circuit is not closed, the call is rejected with CircuitOpenError:

from interlock import CircuitOpenError

try:
    breaker.call(gateway.charge, 100)
except CircuitOpenError as exc:
    # exc.breaker_name, exc.retry_after (seconds, may be None), exc.last_failure
    raise

Inspect state

breaker.state  # State.CLOSED / OPEN / HALF_OPEN / ...
breaker.snapshot()  # WindowSnapshot: total_calls, failed_calls, slow_calls,
# .failure_rate, .slow_call_rate

Next steps