Runnable demo¶

Three self-contained scripts in examples/ show a breaker doing its job — standard library plus interlock-cb only, no network, no services to stand up. The output is deterministic: what you see below is exactly what you get, so set a breakpoint anywhere and step through.

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

python examples/lifecycle.py      # one breaker through its full state cycle
python examples/two_clients.py    # two clients, one outage, no collateral damage
python examples/pipeline.py       # timeout + breaker + fallback around a hanging service

Both scripts print through an EventListener — every line tagged [listener] comes from the breaker itself, not from the demo code. See Observability for the hook reference.

`lifecycle.py` — one breaker, full cycle¶

A fake payment gateway is healthy, goes down, and recovers. The breaker is configured tightly so the whole story fits in seven calls: trip at 50% failures over a 4-call window, stay open 1 second, close after 2 good probes.

lifecycle.py — full source

"""Walk one breaker through its full lifecycle: CLOSED -> OPEN -> HALF_OPEN -> CLOSED.

Zero dependencies, no network — run it directly and watch every transition:

    python examples/lifecycle.py

The gateway is deterministic (healthy, then down, then recovered), so the
output is always the same. Tweak the ``Config`` values and re-run to see how
the thresholds change the story. Explained line by line in
https://bagowix.github.io/interlock/demo/.
"""

import time

from interlock import CircuitBreaker, CircuitOpenError, Config, Outcome, State


class GatewayError(Exception):
    """The fake dependency's failure mode."""


class FlakyGateway:
    """A payment gateway you can switch between healthy and down."""

    def __init__(self) -> None:
        self.healthy = True

    def charge(self, amount: int) -> str:
        if not self.healthy:
            raise GatewayError('503 from gateway')

        return f'charged ${amount}'


class PrintListener:
    """An EventListener that narrates everything the breaker does."""

    def on_state_change(self, *, name: str, old: State, new: State) -> None:
        print(f'  [listener] {name}: state {old.name} -> {new.name}')

    def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
        print(f'  [listener] {name}: recorded {outcome.name} ({duration:.3f}s)')

    def on_rejected(self, *, name: str) -> None:
        print(f'  [listener] {name}: call rejected — circuit is open')

    def on_reset(self, *, name: str) -> None:
        print(f'  [listener] {name}: manual reset')


gateway = FlakyGateway()
breaker = CircuitBreaker(
    name='payments-api',
    config=Config(
        failure_rate_threshold=0.5,  # trip at >= 50% failures ...
        minimum_number_of_calls=4,  # ... once the window holds 4 calls
        window_size=4,
        wait_duration_in_open=1.0,  # stay OPEN for 1s, then allow probes
        permitted_calls_in_half_open=2,  # 2 good probes close the circuit
    ),
    listener=PrintListener(),
)


@breaker
def charge(amount: int) -> str:
    """The protected call: the decorator preserves this signature."""
    return gateway.charge(amount)


def attempt(number: int, amount: int) -> None:
    """Make one charge attempt and print what came back."""
    try:
        print(f'call {number}: {charge(amount)}')
    except GatewayError as exc:
        print(f'call {number}: dependency failed — {exc}')
    except CircuitOpenError as exc:
        print(
            f'call {number}: REJECTED in ~0ms, the gateway was never called '
            f'(retry_after={exc.retry_after:.1f}s)'
        )


def main() -> None:
    """Run the four phases of the lifecycle."""
    print('phase 1 — healthy dependency, breaker CLOSED')
    attempt(1, 10)
    attempt(2, 20)

    print()
    print('phase 2 — the gateway goes down; failures fill the window')
    gateway.healthy = False
    attempt(3, 30)
    attempt(4, 40)  # 2 failures / 4 calls = 50% -> the breaker trips

    print()
    print('phase 3 — circuit OPEN: calls fail fast, the gateway gets a break')
    attempt(5, 50)

    print()
    print('phase 4 — after wait_duration_in_open the breaker probes')
    gateway.healthy = True  # ops fixed the gateway meanwhile
    time.sleep(1.1)
    attempt(6, 60)  # probe 1 (OPEN -> HALF_OPEN on admission)
    attempt(7, 70)  # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED

    print()
    print(f'final state: {breaker.state.name}, window reset: {breaker.snapshot()}')


if __name__ == '__main__':
    main()

Running it prints:

phase 1 — healthy dependency, breaker CLOSED
  [listener] payments-api: recorded SUCCESS (0.000s)
call 1: charged $10
  [listener] payments-api: recorded SUCCESS (0.000s)
call 2: charged $20

Every completed call is recorded into the sliding window — the listener's on_call hook fires with the classified Outcome and the duration.

phase 2 — the gateway goes down; failures fill the window
  [listener] payments-api: recorded FAILURE (0.000s)
call 3: dependency failed — 503 from gateway
  [listener] payments-api: recorded FAILURE (0.000s)
  [listener] payments-api: state CLOSED -> OPEN
call 4: dependency failed — 503 from gateway

Call 4 is the moment the window reaches minimum_number_of_calls=4 with 2 failures out of 4 — exactly the 50% threshold — so recording it trips the circuit: on_state_change fires inside call 4's bookkeeping, before the demo's own call 4: line prints.

phase 3 — circuit OPEN: calls fail fast, the gateway gets a break
  [listener] payments-api: call rejected — circuit is open
call 5: REJECTED in ~0ms, the gateway was never called (retry_after=1.0s)

Call 5 never reaches the gateway. CircuitOpenError is raised immediately and carries retry_after — the estimate until the next probe is allowed. This is the entire point of a breaker: while the dependency is down, callers spend no timeouts on it and it gets quiet time to recover.

phase 4 — after wait_duration_in_open the breaker probes
  [listener] payments-api: state OPEN -> HALF_OPEN
  [listener] payments-api: recorded SUCCESS (0.000s)
call 6: charged $60
  [listener] payments-api: recorded SUCCESS (0.000s)
  [listener] payments-api: state HALF_OPEN -> CLOSED
call 7: charged $70

final state: CLOSED, window reset: WindowSnapshot(total_calls=0, failed_calls=0, slow_calls=0)

The transition to HALF_OPEN is lazy: it happens when call 6 asks for admission after the 1-second wait, not on a timer (set Config.auto_transition=True for the eager variant). Calls 6 and 7 are the two permitted probes; both succeed, so recording the second one closes the circuit and starts a fresh window.

Things to try: set gateway.healthy = False before phase 4 and watch the failed probe reopen the circuit; raise permitted_calls_in_half_open; put a breakpoint in PrintListener.on_state_change and inspect breaker.snapshot() at each transition.

`two_clients.py` — isolation under a partial outage¶

One asyncio app talks to two dependencies. A Registry hands each its own breaker, so when recommendations goes down during rounds 3–6, only its circuit opens — the app serves a cached fallback — while payments keeps charging as if nothing happened.

two_clients.py — full source

"""Two guarded clients in one event loop: one dependency dies, the other keeps serving.

Zero dependencies, no network — run it directly:

    python examples/two_clients.py

A ``Registry`` hands out one independent breaker per dependency. The
``recommendations`` service goes down during rounds 3-6: its breaker opens
and the app falls back to a cached list, while ``payments`` — its own
breaker untouched — keeps charging without a hiccup. Explained line by line
in https://bagowix.github.io/interlock/demo/.
"""

import asyncio

from interlock import CircuitOpenError, Config, Outcome, Registry, State


class RecsDownError(Exception):
    """The recommendations service's failure mode."""


class PrintListener:
    """An EventListener that narrates state changes and rejections."""

    def on_state_change(self, *, name: str, old: State, new: State) -> None:
        print(f'  [listener] {name}: state {old.name} -> {new.name}')

    def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
        pass  # per-call noise is off; see examples/lifecycle.py for it

    def on_rejected(self, *, name: str) -> None:
        print(f'  [listener] {name}: call rejected — circuit is open')

    def on_reset(self, *, name: str) -> None:
        print(f'  [listener] {name}: manual reset')


registry = Registry(
    config=Config(
        failure_rate_threshold=0.5,
        minimum_number_of_calls=4,
        window_size=4,
        wait_duration_in_open=1.0,
        permitted_calls_in_half_open=2,
    ),
    listener=PrintListener(),
)

outage = False


@registry.get('recommendations')
async def fetch_recommendations(user: str) -> list[str]:
    """Call the flaky recommendations service."""
    if outage:
        raise RecsDownError('recommendations service timed out')
    return [f'{user}-pick-1', f'{user}-pick-2']


@registry.get('payments')
async def charge(user: str, amount: int) -> str:
    """Call the payments service, which stays healthy throughout."""
    return f'charged {user} ${amount}'


async def round_trip(number: int, user: str) -> None:
    """One request round: hit both dependencies concurrently."""
    print(f'round {number}:')
    payment, recs = await asyncio.gather(
        charge(user, 25),
        fetch_recommendations(user),
        return_exceptions=True,
    )
    print(f'  payments        -> {payment}')
    if isinstance(recs, CircuitOpenError):
        print('  recommendations -> rejected instantly -> fallback: cached picks')
    elif isinstance(recs, RecsDownError):
        print(f'  recommendations -> failed ({recs}) -> fallback: cached picks')
    else:
        print(f'  recommendations -> {recs}')


async def main() -> None:
    """Run eight rounds across the outage and the recovery."""
    global outage  # noqa: PLW0603 - a module-level switch keeps the demo flat

    print('rounds 1-2 — both dependencies healthy')
    await round_trip(1, 'alice')
    await round_trip(2, 'bob')

    print()
    print('rounds 3-6 — recommendations goes down; payments must not care')
    outage = True
    await round_trip(3, 'carol')
    await round_trip(4, 'dave')  # 50% failures -> recommendations breaker opens
    await round_trip(5, 'erin')  # rejected in ~0ms: no timeout is burned
    await round_trip(6, 'frank')

    print()
    print('rounds 7-8 — the outage is over; the breaker probes and closes')
    outage = False
    await asyncio.sleep(1.1)  # let wait_duration_in_open elapse
    await round_trip(7, 'grace')
    await round_trip(8, 'heidi')

    print()
    for name in ('payments', 'recommendations'):
        print(f'final state of {name}: {registry.get(name).state.name}')


if __name__ == '__main__':
    asyncio.run(main())

The interesting part of the output:

round 4:
  [listener] recommendations: state CLOSED -> OPEN
  payments        -> charged dave $25
  recommendations -> failed (recommendations service timed out) -> fallback: cached picks
round 5:
  [listener] recommendations: call rejected — circuit is open
  payments        -> charged erin $25
  recommendations -> rejected instantly -> fallback: cached picks

Round 4's failure is the second in a 4-call window — the recommendations breaker trips. From round 5 on the difference matters: the round-4 request waited for the dependency to fail, the round-5 request is rejected in microseconds, so the user still gets their page (with cached picks) at full speed. Notice what is absent: the payments breaker never logs a single state change for the whole run.

rounds 7-8 — the outage is over; the breaker probes and closes
round 7:
  [listener] recommendations: state OPEN -> HALF_OPEN
  payments        -> charged grace $25
  recommendations -> ['grace-pick-1', 'grace-pick-2']
round 8:
  [listener] recommendations: state HALF_OPEN -> CLOSED
  payments        -> charged heidi $25
  recommendations -> ['heidi-pick-1', 'heidi-pick-2']

Recovery is gradual by design: two successful probes (rounds 7 and 8) must complete before the circuit closes again.

Things to try: extend the outage past round 7 and watch a failed probe send the circuit straight back to OPEN; give the two breakers different configs via registry.get(name, config=...); replace the hand-rolled fallback with a tenacity retry that waits exactly retry_after.

`pipeline.py` — composition against a quiet death¶

The nastiest failure mode gets the v2 pipeline treatment: the inventory service never raises — it hangs. On its own that trips nothing and starves every caller. Three composed strategies turn it into a non-event: a timeout makes hangs classifiable, the breaker counts them, a fallback keeps serving.

pipeline.py — full source

"""Compose timeout + breaker + fallback around one dependency that dies quietly.

Zero dependencies, no network — run it directly:

    python examples/pipeline.py

The inventory service never raises: it *hangs*. Alone, that starves callers
and trips nothing. The pipeline turns the hang into a failure chain: the
timeout cancels the attempt, the breaker counts it and opens, the fallback
serves a cached snapshot — and while the circuit is open, requests cost
~0.0s instead of a timeout each. Add `.retry(...)` between the fallback and
the breaker for the full stack (needs `interlock-cb[tenacity]`). Explained
line by line in https://bagowix.github.io/interlock/demo/.
"""

import asyncio
import time

from interlock import (
    CallTimeoutError,
    CircuitBreaker,
    CircuitOpenError,
    Config,
    Outcome,
    Pipeline,
    State,
)

CACHED_SNAPSHOT = ['widget (cached)', 'gadget (cached)']


class PrintListener:
    """Narrates the breaker's transitions and the pipeline's decisions."""

    def on_state_change(self, *, name: str, old: State, new: State) -> None:
        print(f'  [listener] {name}: state {old.name} -> {new.name}')

    def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
        pass  # per-call noise is off; see examples/lifecycle.py for it

    def on_rejected(self, *, name: str) -> None:
        print(f'  [listener] {name}: call rejected — circuit is open')

    def on_reset(self, *, name: str) -> None:
        print(f'  [listener] {name}: manual reset')

    def on_fallback(self, *, name: str, error: BaseException) -> None:
        print(f'  [listener] {name}: fallback served instead of {type(error).__name__}')


events = PrintListener()
hanging = False
breaker = CircuitBreaker(
    name='inventory',
    config=Config(
        failure_rate_threshold=0.5,  # trip at >= 50% failures ...
        minimum_number_of_calls=4,  # ... once the window holds 4 calls
        window_size=4,
        wait_duration_in_open=1.0,  # stay OPEN for 1s, then allow probes
        permitted_calls_in_half_open=2,  # 2 good probes close the circuit
    ),
    listener=events,
)

pipeline = (
    Pipeline.builder()
    .fallback(
        lambda _exc: CACHED_SNAPSHOT,
        on=(CircuitOpenError, CallTimeoutError),
        name='inventory',
        listener=events,
    )
    .circuit_breaker(breaker)
    .timeout(0.2)  # a hanging attempt becomes CallTimeoutError after 0.2s
    .build()
)


@pipeline
async def fetch_inventory() -> list[str]:
    """The protected call: the dependency hangs instead of erroring."""
    if hanging:
        await asyncio.sleep(5)  # never finishes within the timeout
    return ['widget', 'gadget']


async def request(number: int) -> None:
    """Serve one request and print what it cost."""
    start = time.perf_counter()
    items = await fetch_inventory()
    elapsed = time.perf_counter() - start
    print(f'request {number}: {items} in ~{elapsed:.1f}s')


async def main() -> None:
    """Run the outage story end to end."""
    global hanging  # noqa: PLW0603 - a module-level switch keeps the demo flat

    print('phase 1 — healthy and fast, breaker CLOSED')
    await request(1)
    await request(2)

    print()
    print('phase 2 — the service starts hanging; timeouts become failures')
    hanging = True
    await request(3)  # waits the full 0.2s timeout, then serves the cache
    await request(4)  # 2 timeouts / 4 calls = 50% -> the breaker trips

    print()
    print('phase 3 — circuit OPEN: no timeout is burned, the cache is instant')
    await request(5)

    print()
    print('phase 4 — the service recovers; probes close the circuit')
    hanging = False
    await asyncio.sleep(1.1)  # let wait_duration_in_open elapse
    await request(6)  # probe 1 (OPEN -> HALF_OPEN on admission)
    await request(7)  # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED

    print()
    print(f'final state: {breaker.state.name}')


if __name__ == '__main__':
    asyncio.run(main())

The interesting part of the output:

phase 2 — the service starts hanging; timeouts become failures
  [listener] inventory: fallback served instead of CallTimeoutError
request 3: ['widget (cached)', 'gadget (cached)'] in ~0.2s
  [listener] inventory: state CLOSED -> OPEN
  [listener] inventory: fallback served instead of CallTimeoutError
request 4: ['widget (cached)', 'gadget (cached)'] in ~0.2s

phase 3 — circuit OPEN: no timeout is burned, the cache is instant
  [listener] inventory: call rejected — circuit is open
  [listener] inventory: fallback served instead of CircuitOpenError
request 5: ['widget (cached)', 'gadget (cached)'] in ~0.0s

Watch the latency column. Requests 3–4 each pay the full 0.2 s timeout — that is the detection cost while the window fills. Request 4 tips the failure rate to 50% and the circuit opens; from request 5 on the rejection is immediate and the cached snapshot costs ~0.0 s. The user never saw an error: every response during the outage came from the fallback, and the listener logged each substitution.

phase 4 — the service recovers; probes close the circuit
  [listener] inventory: state OPEN -> HALF_OPEN
request 6: ['widget', 'gadget'] in ~0.0s
  [listener] inventory: state HALF_OPEN -> CLOSED
request 7: ['widget', 'gadget'] in ~0.0s

Things to try: raise wait_duration_in_open and watch how long the cache serves; put a .retry(...) step between the fallback and the breaker (needs interlock-cb[tenacity]) and see attempts in the listener via on_retry; drop the .timeout(...) step and watch the outage become invisible again — no strategy ever fires.

Where to go next¶

These demos hand-roll their guarded clients so you can debug the mechanics. In real code you usually don't have to: the integrations apply the same per-dependency pattern to httpx2, httpx, aiohttp, requests and FastAPI transparently, and the resilience pipeline composes the breaker with timeout, bulkhead, retry and fallback declaratively.

Runnable demo¶

lifecycle.py — one breaker, full cycle¶

two_clients.py — isolation under a partial outage¶

pipeline.py — composition against a quiet death¶

Where to go next¶

`lifecycle.py` — one breaker, full cycle¶

`two_clients.py` — isolation under a partial outage¶

`pipeline.py` — composition against a quiet death¶