interlock

A modern circuit breaker for Python — sync and async in a single class, sliding-window rate and slow-call detection, a type-safe API, and transparent integrations at the transport level.

uv add interlock-cb          # or: pip install interlock-cb

interlock-cb supports Python 3.11 and newer. The core uses only the standard library; external integrations are installed as optional extras.

Create one named breaker per dependency and reuse it around every call to it:

from interlock import CircuitBreaker, CircuitOpenError, Config

payments = CircuitBreaker(
    name='payments',
    config=Config(
        failure_rate_threshold=0.5,  # trip at 50% failures...
        minimum_number_of_calls=20,  # ...once the window holds 20 calls
        slow_call_duration_threshold=2.0,  # a call slower than 2s counts as slow
        slow_call_rate_threshold=0.3,  # 30% slow calls trip it just as well
    ),
)


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


try:
    receipt = charge(100)
except CircuitOpenError as exc:
    print(exc)  # Circuit 'payments' is open; retry in ~60.000s

The slow-call thresholds matter as much as the failure ones: a dependency that answers every call in 30 seconds raises nothing, so a consecutive-failure counter keeps the circuit closed while your own request queue fills up.

The same instance protects async callables — there is no second class to configure and no separate state to reason about:

@payments
async def refund(charge_id: str) -> None:
    await gateway.refund(charge_id)

The decorator preserves the wrapped signature and whether it is sync or async. breaker.call(fn, ...), with breaker and async with breaker protect the same call in other shapes — see Getting started for all calling styles and Configuration for every threshold.

• Sync and async, one class. CircuitBreaker dispatches to separate sync and async paths without duplicating the public API.
• Failure rates over sliding windows. Choose count- or time-based windows instead of relying only on consecutive failures.
• Slow calls and returned values count. Detect latency degradation and classify unsuccessful results even when no exception is raised.
• Type-safe decorators. Wrapped signatures and their sync/async nature are preserved; the package ships py.typed and passes three strict type checkers.
• Zero-dependency core. Optional clients, frameworks, storage and observability integrations never leak into the core package.
• Composable resilience. Combine timeout, bulkhead, breaker, retry and fallback explicitly, or coordinate breaker state across instances with Redis.

Start a new integration in METRICS_ONLY to observe real failure and slow-call rates without rejecting traffic. The initial state is applied before a lazy per-host breaker can admit its first request:

import httpx2

from interlock import Config, LoggingEventListener, State
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx2.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    listener=LoggingEventListener(),
)

LoggingEventListener writes every event through stdlib logging; swap it for an EventListener that exports to your metrics backend. Hosts are only known at runtime, so transport.registry.items() lists every breaker created so far and get_existing(host) inspects one without creating it. After tuning thresholds, deploy a new transport with the default initial_state=State.CLOSED; the enforcing instance starts with a fresh window. See States and manual control.

A local breaker only reacts to what its own process saw. Back it with Redis and the whole fleet backs off together:

import redis

from interlock import CircuitBreaker
from interlock.integrations.redis import RedisStorage

payments = CircuitBreaker(
    name='payments',
    storage=RedisStorage(redis.Redis(host='redis.internal')),
)

Tripping is atomic across racing instances, recovery probes are budgeted globally rather than per process, and a Redis outage degrades to local state instead of failing calls. Sharing state gates traffic everywhere at once — that is the point, and the risk, so the Redis integration page starts with when not to use it.

Compose strategies in an explicit order (first is outermost) while keeping the breaker useful as a standalone primitive:

from interlock import CircuitBreaker, CircuitOpenError, Pipeline

breaker = CircuitBreaker(name='recommendations')

pipeline = (
    Pipeline.builder()
    .fallback(lambda exc: [], on=(CircuitOpenError,))
    .retry(attempts=4)  # requires interlock-cb[tenacity]
    .circuit_breaker(breaker)
    .bulkhead(8)
    .timeout(2.0)
    .build()
)


@pipeline
async def fetch_picks(user: str) -> list[str]:
    return await client.get_picks(user)

Retries never hammer an open circuit, one hung attempt cannot eat the retry budget, and every decision is observable — see the pipeline guide.

The httpx2 transport applies one breaker per host with no decorators at call sites:

import httpx2
from interlock.integrations.httpx2 import CircuitBreakerTransport

transport = CircuitBreakerTransport(httpx2.HTTPTransport())
client = httpx2.Client(transport=transport)

By default, transport exceptions and the canonical retryable statuses (429, 500, 502, 503, 504) count as failures; 4xx client errors do not.

The integrations overview also includes recipes for LLM SDKs and Flask/Django.

interlock-cb is young: its first release was in 2026. Established libraries such as pybreaker and circuitbreaker have carried production traffic for years and remain a better fit when maturity matters more than the feature differences.

The full comparison covers more features as well as aiobreaker and purgatory. Something out of date or unfair? Please open a PR.

The reliability work compensating for the project's shorter production history includes 100% branch coverage, three strict type checkers, mutation testing of the state machine and engine, property- and model-based tests, and CI on free-threaded CPython. The correctness and testing page documents what is verified and where the limits are.

The full documentation is hosted at https://bagowix.github.io/interlock/. Start with:

• Getting started
• Configuration and states
• Timeouts and retries
• Resilience pipeline
• Integrations
• Correctness and testing
• API reference

For a deterministic, network-free demonstration of every state transition, run the examples/ scripts or follow the walkthrough.

Bug reports and pull requests are welcome. See CONTRIBUTING.md for the local setup and the checks a change must pass, and CODE_OF_CONDUCT.md for community expectations. Security issues: please follow SECURITY.md.

interlock is released under the MIT License.