Interlock

https://github.com/bagowix/interlock
Admin
A modern circuit breaker for Python: sync + async in one class, sliding-window rate, slow-call...
Tokens:159,912
Snippets:1,214
Trust Score:7.5
License:MIT
Update:4 days ago
Show doc for...
Tokens:
Raw
### Install and run interlock-cb examples

Source: https://github.com/bagowix/interlock/blob/main/examples/README.md

Commands to install the library and execute the provided demonstration scripts.

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

python examples/lifecycle.py    # one breaker: CLOSED -> OPEN -> HALF_OPEN -> CLOSED
python examples/two_clients.py  # two guarded clients: one fails, the other keeps serving
python examples/pipeline.py     # timeout + breaker + fallback around a hanging service
```

--------------------------------

### Install and run the interlock-cb examples

Source: https://github.com/bagowix/interlock/blob/main/docs/demo.md

Install the interlock-cb package and run the three self-contained example scripts. The output is deterministic, so you can set breakpoints and step through the code.

```bash
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
```

--------------------------------

### Install and run interlock-cb examples

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Commands to install the package and execute the provided demonstration scripts.

```bash
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
```

--------------------------------

### Install interlock-cb[httpx] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use uv to install the httpx integration package.

```bash
uv add 'interlock-cb[httpx]'
```

--------------------------------

### Install OpenTelemetry extra

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the necessary dependency for OpenTelemetry support using uv.

```bash
uv add 'interlock-cb[otel]'
```

--------------------------------

### Install interlock-cb[tenacity] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Install the tenacity integration using poetry.

```bash
poetry add 'interlock-cb[tenacity]'
```

--------------------------------

### Install interlock-cb[fastapi] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/fastapi.md

Install the FastAPI integration using poetry.

```bash
poetry add 'interlock-cb[fastapi]'
```

--------------------------------

### Install interlock-cb[httpx] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use poetry to install the httpx integration package.

```bash
poetry add 'interlock-cb[httpx]'
```

--------------------------------

### Clone and set up the interlock repository with uv

Source: https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md

Clones the repository, enters the directory, and runs uv sync to create the virtual environment and install development dependencies plus all extras. Requires uv to be installed.

```bash
git clone https://github.com/bagowix/interlock
cd interlock
uv sync            # creates the venv and installs dev + all extras
```

--------------------------------

### Install interlock-cb[tenacity] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Install the tenacity integration using the uv package manager.

```bash
uv add 'interlock-cb[tenacity]'
```

--------------------------------

### Install interlock-cb[redis] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use the uv package manager to install the redis extra for interlock.

```bash
uv add 'interlock-cb[redis]'
```

--------------------------------

### Install interlock-cb[fastapi] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/fastapi.md

Install the FastAPI integration using the uv package manager.

```bash
uv add 'interlock-cb[fastapi]'
```

--------------------------------

### Install optional interlock-cb integrations

Source: https://github.com/bagowix/interlock/blob/main/docs/getting-started.md

Install specific extras for framework or library support using uv, pip, or poetry.

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

--------------------------------

### Install interlock-cb[tenacity] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Install the tenacity integration using pip.

```bash
pip install 'interlock-cb[tenacity]'
```

--------------------------------

### Install interlock-cb[aiohttp] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the aiohttp integration using the uv package manager.

```bash
uv add 'interlock-cb[aiohttp]'
```

--------------------------------

### Install interlock-cb[httpx2] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx2.md

Install the interlock-cb[httpx2] extra using the uv package manager.

```bash
uv add 'interlock-cb[httpx2]'
```

--------------------------------

### Install interlock with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/index.md

Use pip to install the interlock package.

```bash
pip install interlock-cb
```

--------------------------------

### Install interlock-cb[httpx2] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx2.md

Install the interlock-cb[httpx2] extra using poetry.

```bash
poetry add 'interlock-cb[httpx2]'
```

--------------------------------

### Install interlock-cb with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/getting-started.md

Use pip to install the interlock-cb package.

```bash
pip install interlock-cb
```

--------------------------------

### Install interlock-cb[httpx] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use pip to install the httpx integration package.

```bash
pip install 'interlock-cb[httpx]'
```

--------------------------------

### Install interlock-cb[fastapi] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/fastapi.md

Install the FastAPI integration using pip.

```bash
pip install 'interlock-cb[fastapi]'
```

--------------------------------

### Install interlock-cb[requests] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use uv to add the requests integration package.

```bash
uv add 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/getting-started.md

Use uv to add the interlock-cb package to your project.

```bash
uv add interlock-cb
```

--------------------------------

### Install interlock-cb[httpx2] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx2.md

Install the interlock-cb[httpx2] extra using pip.

```bash
pip install 'interlock-cb[httpx2]'
```

--------------------------------

### Install interlock-cb[aiohttp] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the aiohttp integration using the poetry package manager.

```bash
poetry add 'interlock-cb[aiohttp]'
```

--------------------------------

### Install interlock-cb[requests] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use pip to install the requests integration package.

```bash
pip install 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb[redis] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use pip to install the redis extra for interlock.

```bash
pip install 'interlock-cb[redis]'
```

--------------------------------

### Install interlock-cb[requests] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/requests.md

Install the requests extra using uv. This adds the necessary dependencies for the requests integration.

```bash
uv add 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb[otel] extra

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/observability.md

Installs the OpenTelemetry extra package using uv. This adds the OTel listener while keeping the core dependency-free.

```bash
uv add 'interlock-cb[otel]'
```

--------------------------------

### Install interlock with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/index.md

Use the uv package manager to add interlock to your project.

```bash
uv add interlock-cb
```

--------------------------------

### Install interlock-cb[tenacity] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/tenacity.md

Adds the interlock-cb[tenacity] extra using uv.

```bash
uv add 'interlock-cb[tenacity]'
```

--------------------------------

### Install interlock-cb[aiohttp] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/aiohttp.md

Adds the aiohttp extra using the uv package manager.

```bash
uv add 'interlock-cb[aiohttp]'
```

--------------------------------

### Install interlock-cb[litestar] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Adds the litestar extra to the project using uv. Requires uv to be installed.

```bash
uv add 'interlock-cb[litestar]'
```

--------------------------------

### Install interlock-cb with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/getting-started.md

Use poetry to add the interlock-cb package to your project.

```bash
poetry add interlock-cb
```

--------------------------------

### Install the pre-commit git hook with prek

Source: https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md

One-time command to install the git hook that runs the fast subset of checks automatically.

```bash
uv run prek install           # one-time, installs the git hook
```

--------------------------------

### Install interlock-cb[requests] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use poetry to add the requests integration package.

```bash
poetry add 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb[aiohttp] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the aiohttp integration using the pip package manager.

```bash
pip install 'interlock-cb[aiohttp]'
```

--------------------------------

### Install interlock-cb[litestar] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the litestar extra using pip. Requires pip to be available.

```bash
pip install 'interlock-cb[litestar]'
```

--------------------------------

### Install interlock-cb[requests] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/requests.md

Install the requests extra using poetry. This adds the necessary dependencies for the requests integration.

```bash
poetry add 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb[httpx2] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Installs the httpx2 extra using pip. The extra wraps httpx2 transports with a per-host circuit breaker.

```bash
pip install 'interlock-cb[httpx2]'
```

--------------------------------

### Install interlock-cb[litestar] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Adds the litestar extra to the project using poetry. Requires poetry to be installed.

```bash
poetry add 'interlock-cb[litestar]'
```

--------------------------------

### Install interlock-cb[aiohttp] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/aiohttp.md

Adds the aiohttp extra using poetry.

```bash
poetry add 'interlock-cb[aiohttp]'
```

--------------------------------

### Async RedisStorage setup with async redis client

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/redis.md

Shows how to create an AsyncRedisStorage with an async redis client and pass it to a CircuitBreaker. The storage serves only the async API (async with, async call). Mixing sync and async styles raises InterlockError.

```python
import redis.asyncio
from interlock import CircuitBreaker
from interlock.integrations.redis import AsyncRedisStorage

storage = AsyncRedisStorage(redis.asyncio.Redis(host='redis.internal'))
breaker = CircuitBreaker(name='payments', storage=storage)
```

--------------------------------

### Install interlock-cb[redis] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Use poetry to add the redis extra for interlock.

```bash
poetry add 'interlock-cb[redis]'
```

--------------------------------

### Install interlock-cb[aiohttp] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/aiohttp.md

Adds the aiohttp extra using pip.

```bash
pip install 'interlock-cb[aiohttp]'
```

--------------------------------

### two_clients.py full source: two guarded clients with independent breakers

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Complete runnable example (python examples/two_clients.py) with zero dependencies. It creates a Registry with a 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) and a PrintListener. Two services are registered: 'recommendations' (flaky, raises RecsDownError when outage is True) and 'payments' (always healthy). The main() runs eight rounds: rounds 1-2 healthy, rounds 3-6 outage (recommendations fails, breaker opens, calls are rejected instantly), rounds 7-8 after outage (breaker probes and closes). Output shows state changes and fallback behavior.

```python
"""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())
    
```

--------------------------------

### Install interlock-cb[requests] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/requests.md

Install the requests extra using pip. This adds the necessary dependencies for the requests integration.

```bash
pip install 'interlock-cb[requests]'
```

--------------------------------

### Install interlock-cb[litestar] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/litestar.md

Adds the Litestar integration extra using the uv package manager.

```bash
uv add 'interlock-cb[litestar]'
```

--------------------------------

### Install interlock-cb[httpx2] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Adds the httpx2 extra to a project using uv. The extra wraps httpx2 transports with a per-host circuit breaker.

```bash
uv add 'interlock-cb[httpx2]'
```

--------------------------------

### Install interlock-cb with uv or pip

Source: https://github.com/bagowix/interlock/blob/main/README.md

Install the interlock-cb package using uv or pip. Requires Python 3.11 or newer; the core uses only the standard library, with external integrations available as optional extras.

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

--------------------------------

### Run Redis-backed tests with a real server

Source: https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md

Starts a Redis (or Valkey) container and runs the test suite with INTERLOCK_TEST_REDIS_URL set. Required for concurrency tests that assert atomicity only a real server guarantees; otherwise fakeredis is used by default.

```bash
docker run --rm -p 6379:6379 redis    # or valkey/valkey, or a local redis-server
INTERLOCK_TEST_REDIS_URL=redis://localhost:6379/0 uv run pytest
```

--------------------------------

### Install interlock-cb[litestar] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/litestar.md

Adds the Litestar integration extra using Poetry.

```bash
poetry add 'interlock-cb[litestar]'
```

--------------------------------

### Constructor call with keyword arguments (bad vs good)

Source: https://github.com/bagowix/interlock/blob/main/AGENTS.md

Shows the project's rule that calls with 3+ arguments must use keyword arguments only. The 'bad' example uses positional arguments; the 'good' example uses keyword arguments.

```python
# bad
config = Config(0.5, 10, 60.0)
```

```python
# good
config = Config(failure_rate_threshold=0.5, minimum_number_of_calls=10, wait_duration_in_open=60.0)
```

--------------------------------

### Sync RedisStorage setup with CircuitBreaker and Registry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/redis.md

Shows how to create a RedisStorage with a sync redis client and pass it to a CircuitBreaker or a Registry. A Registry hands the storage to every breaker it creates, each coordinating under its own name. The storage serves only the sync API (with, sync call).

```python
import redis
from interlock import CircuitBreaker, Registry
from interlock.integrations.redis import RedisStorage

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

registry = Registry(storage=storage)  # or share one storage across many breakers
```

--------------------------------

### Install interlock with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/index.md

Use poetry to add interlock to your project dependencies.

```bash
poetry add interlock-cb
```

--------------------------------

### Install interlock-cb[redis] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/redis.md

Installs the Redis extra using pip. Use this when you want to coordinate breaker state across processes and machines.

```bash
pip install 'interlock-cb[redis]'
```

--------------------------------

### Install interlock-cb[tenacity] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/tenacity.md

Adds the interlock-cb[tenacity] extra using poetry.

```bash
poetry add 'interlock-cb[tenacity]'
```

--------------------------------

### Install interlock-cb[redis] with uv

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/redis.md

Adds the Redis extra to the project using uv. Use this when you want to coordinate breaker state across processes and machines.

```bash
uv add 'interlock-cb[redis]'
```

--------------------------------

### Start in shadow mode with METRICS_ONLY and LoggingEventListener

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx.md

Initializes an async transport with initial_state=State.METRICS_ONLY and a LoggingEventListener to record outcomes without raising CircuitOpenError. Allows inspection via transport.registry.get_existing(host), items(), and names(). Swap listener for an EventListener to export to metrics.

```python
import httpx

from interlock import LoggingEventListener, State
from interlock.integrations.httpx import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    listener=LoggingEventListener(),
)
```

--------------------------------

### Install interlock-cb[litestar] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/litestar.md

Adds the Litestar integration extra using pip.

```bash
pip install 'interlock-cb[litestar]'
```

--------------------------------

### Install interlock-cb[tenacity] with pip

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/tenacity.md

Adds the interlock-cb[tenacity] extra using pip.

```bash
pip install 'interlock-cb[tenacity]'
```

--------------------------------

### two_clients.py — full source

Source: https://github.com/bagowix/interlock/blob/main/docs/demo.md

The full source code for the example script. It is included via a transclusion directive and is not displayed on this page.

```python
--8<-- "examples/two_clients.py"
```

--------------------------------

### Install interlock-cb[httpx2] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Adds the httpx2 extra to a project using poetry. The extra wraps httpx2 transports with a per-host circuit breaker.

```bash
poetry add 'interlock-cb[httpx2]'
```

--------------------------------

### Add interlock Agent Skill with npx

Source: https://github.com/bagowix/interlock/blob/main/README.md

Install the Agent Skill for interlock into coding agents that support the open skills format, such as Claude Code, Cursor, Codex, GitHub Copilot, and Gemini CLI. The skill guides agents through adding circuit breakers, including inventory, integration selection, threshold sizing, shadow-mode rollout, and tests with a fake clock.

```bash
npx skills add bagowix/interlock
```

--------------------------------

### Enable shadow mode and reset breaker

Source: https://github.com/bagowix/interlock/blob/main/docs/migration.md

Use breaker.metrics_only() to observe production without rejecting anything, then read snapshot metrics and call breaker.reset() to start enforcing with a clean window.

```python
breaker.metrics_only()  # observe production, reject nothing
# ... read breaker.snapshot().failure_rate / .slow_call_rate over real traffic ...
breaker.reset()  # start enforcing with a clean window
```

--------------------------------

### Registry constructor and method signatures

Source: https://github.com/bagowix/interlock/blob/main/docs/reference.md

Shows the constructor signature and the available methods on the Registry class. The constructor accepts optional configuration, clock, initial state, classifier, listener, and storage. Methods include get, get_existing, names, items, close_all, and aclose_all.

```python
Registry(*, config=None, clock=None, initial_state=State.CLOSED,
         classifier=None, listener=None, storage=None)
registry.get(name, *, config=None) -> CircuitBreaker
registry.get_existing(name) -> CircuitBreaker | None
registry.names() -> tuple[str, ...]
registry.items() -> tuple[tuple[str, CircuitBreaker], ...]
registry.close_all() / await registry.aclose_all()
```

--------------------------------

### Share Configuration with Registry

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/configuration.md

Initializes a Registry with a default configuration and demonstrates how to apply per-name overrides during breaker creation.

```python
from interlock import Config, Registry, State

registry = Registry(
    config=Config(minimum_number_of_calls=20),
    initial_state=State.METRICS_ONLY,
)

payments = registry.get('payments')  # shared default
search = registry.get('search', config=Config(window_size=500))  # per-name override
```

--------------------------------

### Registry Initialization and Dependency Wrapping

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Demonstrates sharing a Registry instance across the application and wrapping outgoing calls with a breaker.

```python
from interlock import Registry

registry = Registry()
payments = registry.get('payments-api')


def charge(amount: int) -> str:
    return payments.call(gateway.charge, amount)
```

--------------------------------

### Add interlock-cb[httpx] with uv, pip, or poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx.md

Install the httpx extra using your package manager. Supports httpx 0.27.0 and newer.

```bash
uv add 'interlock-cb[httpx]'
```

```bash
pip install 'interlock-cb[httpx]'
```

```bash
poetry add 'interlock-cb[httpx]'
```

--------------------------------

### Using metrics_only() and reset() for shadow mode

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/states.md

Shows how to observe a breaker in production without enforcing thresholds by calling metrics_only(), then reset() to start enforcing with a clean window. The comment indicates inspecting snapshot() until thresholds look right.

```python
breaker.metrics_only()  # observe in production without enforcing
# ... inspect breaker.snapshot() until thresholds look right ...
breaker.reset()  # start enforcing with a clean window
```

--------------------------------

### Configure FastAPI Circuit Breaker

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/fastapi.md

Install the exception handler and inject a breaker dependency into a route.

```python
from typing import Annotated

from fastapi import Depends, FastAPI
from interlock import CircuitBreaker, Registry
from interlock.integrations.fastapi import breaker_dependency, install_exception_handler

app = FastAPI()
registry = Registry()
install_exception_handler(app)

orders_db = breaker_dependency('orders-db', registry=registry)


@app.get('/orders')
async def orders(breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> list[dict]:
    return await breaker.call(fetch_orders)
```

--------------------------------

### Declarative Pipeline Composition

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Example of using the Pipeline builder to compose retry, circuit breaker, and timeout strategies in a specific order.

```python
pipeline = (
    Pipeline.builder()
    .retry(attempts=4)  # retry outside — the recommended order
    .circuit_breaker(breaker)
    .timeout(2.0)
    .build()
)
```

--------------------------------

### registry.close_all() / await registry.aclose_all()

Source: https://github.com/bagowix/interlock/blob/main/docs/reference.md

Closes all breakers created so far. The cache is kept, so get() continues to return the same torn-down instances.

```APIDOC
## registry.close_all() / await registry.aclose_all()

### Description
Closes every breaker created so far. The cache is kept, so get() keeps returning the same, torn-down instances instead of silently starting a fresh lane after shutdown.
```

--------------------------------

### registry.close_all() / await registry.aclose_all()

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Closes every breaker created so far. The cache is kept, so get() continues to return the same torn-down instances.

```APIDOC
## registry.close_all() / await registry.aclose_all()

### Description
Closes every breaker created so far. The cache is kept, so get() continues to return the same torn-down instances.

### Returns
None.
```

--------------------------------

### Safe production rollout with initial_state=State.METRICS_ONLY

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/requests.md

Use this to record real outcomes without rejecting requests during a safe production rollout. The state is applied to every lazily created host before its first request. The public adapter.registry supports local diagnosis with get_existing(host), state, snapshot(), names(), and items(). LoggingEventListener writes every event through stdlib logging; swap it for an EventListener that exports to your metrics backend, then deploy a new adapter with the default CLOSED state to begin enforcement.

```python
from interlock import LoggingEventListener, State
from interlock.integrations.requests import CircuitBreakerAdapter

adapter = CircuitBreakerAdapter(
    initial_state=State.METRICS_ONLY,
    listener=LoggingEventListener(),
)
```

--------------------------------

### Iterate over all breakers with registry.items()

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Example of using registry.items() to enumerate all breakers created so far. It prints the name, state, and snapshot for each breaker. This is the only way to see breakers created lazily by HTTP transports.

```python
for name, breaker in registry.items():
    print(name, breaker.state, breaker.snapshot())
```

--------------------------------

### Initialize Synchronous Circuit Breaker with Redis

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Configures a circuit breaker using the synchronous Redis client and RedisStorage. This setup can be used directly or via a Registry.

```python
import redis
from interlock import CircuitBreaker, Registry
from interlock.integrations.redis import RedisStorage

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

registry = Registry(storage=storage)  # or share one storage across many breakers
```

--------------------------------

### Install interlock-cb[redis] with poetry

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/redis.md

Adds the Redis extra to the project using Poetry. Use this when you want to coordinate breaker state across processes and machines.

```bash
poetry add 'interlock-cb[redis]'
```

--------------------------------

### Fail-fast error handling example

Source: https://github.com/bagowix/interlock/blob/main/AGENTS.md

Shows a bad pattern that hides errors behind a fallback and a good pattern that raises immediately on invalid input. Use the good pattern to fail fast and provide transparent errors.

```python
# bad — the error is hidden behind a fallback
def get_state(name: str) -> State:
    try:
        return storage.load(name)
    except Exception:
        return State.CLOSED


# good — fail fast, transparent error
def get_state(name: str) -> State:
    if not name:
        raise ValueError(f'Empty breaker name: {name!r}')
    return storage.load(name)
```

--------------------------------

### Shadow mode rollout with AsyncCircuitBreakerTransport

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/httpx2.md

Configure an async transport with initial_state=State.METRICS_ONLY and a LoggingEventListener. This starts breakers in metrics-only mode, recording outcomes without raising CircuitOpenError.

```python
import httpx2

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

transport = AsyncCircuitBreakerTransport(
    httpx2.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    listener=LoggingEventListener(),
)
```

--------------------------------

### Configure unreachable exceptions for CircuitBreaker and Registry

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Python example showing how to define a custom exception NoLocalSlot and pass it as unreachable_exceptions to both CircuitBreaker and Registry, so that such failures are treated as inconclusive during HALF_OPEN probes.

```python
from interlock import CircuitBreaker, Registry


class NoLocalSlot(Exception):
    """Raised by the guard's own pool when it has no permit to give out."""


breaker = CircuitBreaker(name='payments', unreachable_exceptions=(NoLocalSlot,))
registry = Registry(unreachable_exceptions=(NoLocalSlot,))
```

--------------------------------

### pipeline.py full source (included via literalinclude)

Source: https://github.com/bagowix/interlock/blob/main/docs/demo.md

The full source of pipeline.py is included via a literalinclude directive, so the actual code is not displayed on this page. The example composes timeout, circuit breaker, and fallback strategies to handle a hanging service.

```python
--8<-- "examples/pipeline.py"
```

--------------------------------

### Safe rollout with shadow mode using AsyncCircuitBreakerTransport

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Configures an async transport in shadow mode (METRICS_ONLY) with a LoggingEventListener. Every host created later starts in METRICS_ONLY before its first request, recording outcomes without raising CircuitOpenError. LoggingEventListener writes events through stdlib logging; swap it for an EventListener that exports to a metrics backend.

```python
import httpx2

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

transport = AsyncCircuitBreakerTransport(
    httpx2.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    listener=LoggingEventListener(),
)
```

--------------------------------

### lifecycle.py output — phase 4: probe and recovery

Source: https://github.com/bagowix/interlock/blob/main/docs/demo.md

Output from lifecycle.py after the open wait. The transition to HALF_OPEN is lazy, happening when call 6 asks for admission after the 1-second wait (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.

```text
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)
```

--------------------------------

### Guard LLM calls with breaker and retries (interlock)

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/llm.md

Shows a complete setup: a breaker with a slow-call threshold (30s) and classifier, a decorated async completion function, and a tenacity retry that stops on circuit open. Uses retry_unless_open and exponential jitter; SDK's own retries should be disabled to avoid multiplying attempts.

```python
import anthropic
from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter

from interlock import CircuitBreaker, Config
from interlock.integrations.tenacity import retry_unless_open

client = anthropic.AsyncAnthropic()

breaker = CircuitBreaker(
    name='anthropic',
    config=Config(slow_call_duration_threshold=30.0),
    classifier=LLMFailureClassifier(),
)


@breaker
async def complete(prompt: str) -> str:
    message = await client.messages.create(
        model='claude-sonnet-5',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}],
    )
    return message.content[0].text


retrying = AsyncRetrying(
    retry=retry_unless_open(
        anthropic.APIStatusError,
        anthropic.APIConnectionError,
        anthropic.APITimeoutError,
    ),
    wait=wait_exponential_jitter(initial=1.0, max=30.0),
    stop=stop_after_attempt(4),
    reraise=True,
)

answer = await retrying(complete, 'Summarise this document...')
```

--------------------------------

### Define a custom Measuring strategy with execute and execute_async

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/pipeline.md

Shows a structural strategy class that times the next layer. The execute method uses time.perf_counter and prints elapsed time, while execute_async awaits the call. The example builds a pipeline with the strategy and a 2.0 second timeout.

```python
import time
from collections.abc import Awaitable, Callable
from typing import TypeVar

T = TypeVar('T')


class Measuring:
    """Times every layer below it."""

    def execute(self, call: Callable[[], T]) -> T:
        start = time.perf_counter()
        try:
            return call()
        finally:
            print(f'took {time.perf_counter() - start:.3f}s')

    async def execute_async(self, call: Callable[[], Awaitable[T]]) -> T:
        start = time.perf_counter()
        try:
            return await call()
        finally:
            print(f'took {time.perf_counter() - start:.3f}s')


pipeline = Pipeline.builder().add(Measuring()).timeout(2.0).build()
```

--------------------------------

### Safe production rollout with initial_state=State.METRICS_ONLY

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/aiohttp.md

Use this to record real outcomes without rejecting requests. The initial_state is applied to every lazily created host before its first request. The public middleware.registry supports local diagnosis with get_existing(host), state, snapshot(), names(), and items(). LoggingEventListener writes every event through stdlib logging; swap it for an EventListener that exports to your metrics backend, then deploy a new middleware with the default CLOSED state to begin enforcement.

```python
from interlock import LoggingEventListener, State
from interlock.integrations.aiohttp import CircuitBreakerMiddleware

middleware = CircuitBreakerMiddleware(
    initial_state=State.METRICS_ONLY,
    listener=LoggingEventListener(),
)
```

--------------------------------

### Create virtual environment with uv

Source: https://github.com/bagowix/interlock/blob/main/AGENTS.md

Creates a Python 3.12 virtual environment in the .venv directory using uv.

```bash
uv venv --python 3.12 .venv
```

--------------------------------

### Share a Registry and wrap outgoing dependencies

Source: https://github.com/bagowix/interlock/blob/main/docs/integrations/frameworks.md

Shows how to create a shared Registry and use it to wrap an outgoing call with a circuit breaker. The handler only translates the rejection; the breakers guard outgoing calls.

```python
from interlock import Registry

registry = Registry()
payments = registry.get('payments-api')


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

```

--------------------------------

### Set up and run tests on free-threaded CPython

Source: https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md

Creates a separate environment (.venv-ft) for free-threaded CPython 3.15t and runs the test suite. Use this to avoid disturbing your main .venv. Note that tests/test_redis_storage.py requires a real Redis server on free-threaded builds because the fakeredis fallback imports lupa, which re-enables the GIL and causes a collection error due to filterwarnings = "error". Set INTERLOCK_TEST_REDIS_URL to avoid importing fakeredis.

```bash
UV_PROJECT_ENVIRONMENT=.venv-ft uv sync --python 3.15t
UV_PROJECT_ENVIRONMENT=.venv-ft uv run pytest
```

--------------------------------

### Output: phases 2-3 (timeouts, circuit open, fallback)

Source: https://github.com/bagowix/interlock/blob/main/docs/demo.md

Output showing the service hanging, timeouts becoming failures, the circuit opening after request 4, and the fallback serving cached data. Requests 3-4 each pay the full 0.2s timeout; from request 5 on, rejection is immediate and the cached snapshot costs ~0.0s.

```text
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
```

--------------------------------

### Initialize Config dataclass

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/configuration.md

Create a new Config instance by providing keyword-only arguments for circuit breaker thresholds and window settings.

```python
from interlock import Config
from interlock import WindowType

config = Config(
    failure_rate_threshold=0.5,
    minimum_number_of_calls=20,
    slow_call_duration_threshold=2.0,
    slow_call_rate_threshold=1.0,
    permitted_calls_in_half_open=10,
    max_concurrent_probes=1,
    wait_duration_in_open=30.0,
    window_type=WindowType.COUNT_BASED,
    window_size=100,
)
```

--------------------------------

### Basic CircuitBreaker usage

Source: https://github.com/bagowix/interlock/blob/main/docs/index.md

Example of decorating a function with a CircuitBreaker and handling CircuitOpenError when the dependency is unhealthy.

```python
from interlock import CircuitBreaker, CircuitOpenError

breaker = CircuitBreaker(name='payments')


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


try:
    charge(100)
except CircuitOpenError as exc:
    ...  # rejected fast: the dependency is unhealthy; retry after exc.retry_after
```

--------------------------------

### SlidingWindow.snapshot()

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Protocol method to get a snapshot of the sliding window state. Implement to provide a WindowSnapshot.

```APIDOC
## SlidingWindow.snapshot()

### Description
Returns a snapshot of the sliding window state. Implement this protocol to provide a WindowSnapshot.

### Method
snapshot() -> WindowSnapshot

### Returns
- **WindowSnapshot** - A snapshot of the sliding window state.
```

--------------------------------

### Circuit Breaker Recovery Log

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Example output showing the state transitions of a circuit breaker during recovery from an outage.

```text
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']
```

--------------------------------

### Clock.monotonic()

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Protocol method to get a monotonic time value. Implement to inject a fake clock for deterministic tests.

```APIDOC
## Clock.monotonic()

### Description
Returns a monotonic time value as a float. Implement this protocol to provide a custom time source, useful for deterministic testing.

### Method
monotonic() -> float

### Returns
- **float** - A monotonic time value.
```

--------------------------------

### FailureClassifier example: OnlyConnectionError

Source: https://github.com/bagowix/interlock/blob/main/docs/migration.md

Shows a custom FailureClassifier that counts only ConnectionError as a failure. This is the interlock replacement for circuitbreaker's expected_exception.

```python
class OnlyConnectionError:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        return isinstance(exception, ConnectionError)
```

--------------------------------

### Registry constructor and method signatures

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

Shows the constructor signature and the main methods of the Registry class. The constructor accepts optional config, clock, initial_state, classifier, listener, and storage. The same name always returns the same instance; a per-call config override applies only at creation. get_existing() inspects the cache without creating a missing name.

```python
Registry(*, config=None, clock=None, initial_state=State.CLOSED,
         classifier=None, listener=None, storage=None)
registry.get(name, *, config=None) -> CircuitBreaker
registry.get_existing(name) -> CircuitBreaker | None
registry.names() -> tuple[str, ...]
registry.items() -> tuple[tuple[str, CircuitBreaker], ...]
registry.close_all() / await registry.aclose_all()
```

--------------------------------

### FailureClassifier example: OnlyConnectionError

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

A FailureClassifier that counts only ConnectionError as a failure. This is the interlock replacement for circuitbreaker's expected_exception, which names the exceptions that do count.

```python
class OnlyConnectionError:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        return isinstance(exception, ConnectionError)
```

--------------------------------

### Registry constructor

Source: https://github.com/bagowix/interlock/blob/main/docs/reference.md

Creates a Registry instance with optional configuration, clock, initial state, classifier, listener, and storage.

```APIDOC
## Registry(*, config=None, clock=None, initial_state=State.CLOSED, classifier=None, listener=None, storage=None)

### Description
Creates a Registry instance that manages named circuit breakers. The same name always returns the same instance; the per-call config override applies only at creation. A storage is handed to every breaker the registry creates.

### Parameters
- **config** (Config) - Optional - Configuration for the registry and created breakers.
- **clock** (Clock) - Optional - Clock for time-based operations.
- **initial_state** (State) - Optional - Initial state for breakers created by the registry. Defaults to State.CLOSED.
- **classifier** (Classifier) - Optional - Classifier for classifying outcomes.
- **listener** (Listener) - Optional - Listener for breaker events.
- **storage** (Storage) - Optional - Storage backend for breakers.
```

--------------------------------

### Run pipeline.py outage simulation

Source: https://github.com/bagowix/interlock/blob/main/docs/llms-full.txt

The main entry point for the outage simulation, demonstrating state transitions in the circuit breaker.

```python
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())
```

--------------------------------

### Code Commenting Standards

Source: https://github.com/bagowix/interlock/blob/main/AGENTS.md

Examples of bad commenting practices that state the obvious versus good practices that explain non-obvious logic and trade-offs.

```python
# bad — states the obvious
# open the circuit
self._state = State.OPEN

# good — explains what is not obvious
# In HALF_OPEN admit at most N probes at a time, otherwise the whole concurrent load
# rushes in as probes and knocks over the barely recovered dependency.

```

--------------------------------

### Default failure classifier usage

Source: https://github.com/bagowix/interlock/blob/main/docs/guides/failure-classification.md

By default, a call is a failure exactly when it raises; any returned value is a success. This example creates a CircuitBreaker with the default classifier.

```python
from interlock import CircuitBreaker

breaker = CircuitBreaker(name='svc')  # DefaultFailureClassifier
```
Tokens:
Raw