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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

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 per dependency and reuse it around every call to it:

frominterlockimportCircuitBreaker, CircuitOpenError, Configpayments=CircuitBreaker(
name='payments',
config=Config(
failure_rate_threshold=0.5, # trip at 50% failures...minimum_number_of_calls=20, # ...once the window holds 20 callsslow_call_duration_threshold=2.0, # a call slower than 2s counts as slowslow_call_rate_threshold=0.3, # 30% slow calls trip it just as well
),
)
@paymentsdefcharge(amount: int) ->str:
returngateway.charge(amount)
try:
receipt=charge(100)
exceptCircuitOpenErrorasexc:
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:

@paymentsasyncdefrefund(charge_id: str) ->None:
awaitgateway.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.

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:

importhttpx2frominterlockimportConfig, LoggingEventListener, Statefrominterlock.integrations.httpx2importAsyncCircuitBreakerTransporttransport=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.

Shared state across instances

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

importredisfrominterlockimportCircuitBreakerfrominterlock.integrations.redisimportRedisStoragepayments=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.

Resilience pipeline

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

frominterlockimportCircuitBreaker, CircuitOpenError, Pipelinebreaker=CircuitBreaker(name='recommendations')
pipeline= (
Pipeline.builder()
.fallback(lambdaexc: [], on=(CircuitOpenError,))
.retry(attempts=4) # requires interlock-cb[tenacity]
.circuit_breaker(breaker)
.bulkhead(8)
.timeout(2.0)
.build()
)
@pipelineasyncdeffetch_picks(user: str) ->list[str]:
returnawaitclient.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:

importhttpx2frominterlock.integrations.httpx2importCircuitBreakerTransporttransport=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.

Integration	Install	Documentation
httpx2	`interlock-cb[httpx2]`	Per-host transport
httpx	`interlock-cb[httpx]`	Per-host transport
aiohttp	`interlock-cb[aiohttp]`	Client middleware
requests	`interlock-cb[requests]`	Session adapter
FastAPI	`interlock-cb[fastapi]`	`503 + Retry-After` handler
Litestar	`interlock-cb[litestar]`	`503 + Retry-After` handler
tenacity	`interlock-cb[tenacity]`	Retry composition
Redis	`interlock-cb[redis]`	Shared state
OpenTelemetry	`interlock-cb[otel]`	Metrics listener

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.

Feature	interlock-cb	pybreaker	circuitbreaker
Core states (closed / open / half-open)	✅	✅	✅
Native asyncio	✅	Tornado	✅
Trip condition	failure rate	consecutive failures	consecutive failures
Time-based sliding window	✅	—	—
Slow-call detection	✅	—	—
Shared state across processes	✅	✅	—
Composable resilience pipeline	✅	—	—
Fully typed API (`py.typed`)	✅	—	—

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:

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.

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
interlock		interlock
overrides		overrides
scripts		scripts
tests		tests
.coderabbit.yaml		.coderabbit.yaml
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
context7.json		context7.json
pyproject.toml		pyproject.toml
uv.lock		uv.lock
zensical.toml		zensical.toml

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
interlock		interlock
overrides		overrides
scripts		scripts
tests		tests
.coderabbit.yaml		.coderabbit.yaml
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
context7.json		context7.json
pyproject.toml		pyproject.toml
uv.lock		uv.lock
zensical.toml		zensical.toml

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
interlock		interlock
overrides		overrides
scripts		scripts
tests		tests
.coderabbit.yaml		.coderabbit.yaml
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
context7.json		context7.json
pyproject.toml		pyproject.toml
uv.lock		uv.lock
zensical.toml		zensical.toml

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
interlock		interlock
overrides		overrides
scripts		scripts
tests		tests
.coderabbit.yaml		.coderabbit.yaml
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
context7.json		context7.json
pyproject.toml		pyproject.toml
uv.lock		uv.lock
zensical.toml		zensical.toml

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
interlock		interlock
overrides		overrides
scripts		scripts
tests		tests
.coderabbit.yaml		.coderabbit.yaml
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
context7.json		context7.json
pyproject.toml		pyproject.toml
uv.lock		uv.lock
zensical.toml		zensical.toml

Latest commit

History

Folders and files

Repository files navigation

interlock

Installation

Quickstart

Why interlock

Safe production rollout

Shared state across instances

Resilience pipeline

Integrations

How it compares

Documentation

Contributing

License

About

Topics

Resources

Code of conduct

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

interlock

Installation

Quickstart

Why interlock

Safe production rollout

Shared state across instances

Resilience pipeline

Integrations

How it compares

Documentation

Contributing

License

About

Topics

Resources

Code of conduct

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

interlock

Installation

Quickstart

Why interlock

Safe production rollout

Shared state across instances

Resilience pipeline

Integrations

How it compares

Documentation

Contributing

License

About

Topics

Resources

Code of conduct