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.

Installation

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.

Quickstart

Create one named breaker and reuse it around calls to the same dependency:

from interlock import CircuitBreaker, CircuitOpenError, Config

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


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


try:
    receipt = charge(100)
except CircuitOpenError as exc:
    print(exc)

The decorator preserves the function's signature and whether it is sync or async. The same breaker also supports breaker.call(fn, ...), with breaker, and async with breaker. See Getting started for all calling styles and Configuration for every threshold.

Why interlock

• 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.

Safe production rollout

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, 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=metrics_listener,
)

Use an EventListener for production metrics. For local diagnostics, transport.registry.get_existing(host) returns an already-created breaker without creating one, so its state and snapshot() can be inspected safely. 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.

Resilience pipeline

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.

Integrations

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.

How it compares

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.

Documentation

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

• Getting started
• Configuration and states
• 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.

Contributing

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.

License

interlock is released under the MIT License.