### Install interlock-cb

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

Install the interlock-cb package using pip or uv.

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

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

### Install Interlock with FastAPI extra

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

Install the necessary package using uv, pip, or poetry.

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

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

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

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

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

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

Use uv to add the interlock-cb[requests] extra for installing the library.

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

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

### Install interlock with uv

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

Install the interlock-cb package using the uv package manager.

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

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

### Install Interlock with Httpx2

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

Install interlock-cb with optional extras for per-host httpx2 transport.

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

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

### Install Interlock with OpenTelemetry

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

Install interlock-cb with optional extras for OpenTelemetry metrics listener.

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

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

### Install httpx2 with Interlock

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

Install the interlock-cb[httpx2] extra using uv, pip, or poetry.

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

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

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

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

### Run Pipeline Example

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

Execute the pipeline.py script to demonstrate a pipeline with timeout, breaker, and fallback around a hanging service.

```bash
python examples/pipeline.py     # timeout + breaker + fallback around a hanging service
```

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

### Install aiohttp integration with poetry

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

Install the aiohttp integration for interlock using poetry.

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

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

### Install aiohttp integration with uv

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

Install the aiohttp integration for interlock using the uv package manager.

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

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

### Install interlock with pip

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

Install the interlock-cb package using pip.

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

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

### Install Interlock CB

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

Install the interlock-cb package using pip or uv.

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

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

### Default Configuration Example

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

Instantiate Config with default values for various parameters like failure rate, minimum calls, and window type.

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

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

### Install Interlock with pip

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

Use pip to install the Interlock library with the FastAPI extra. This command installs the core library and specific components for FastAPI.

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

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

### Run Lifecycle Example

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

Execute the lifecycle.py script to observe the full state cycle of a single circuit breaker.

```bash
python examples/lifecycle.py    # one breaker: CLOSED -> OPEN -> HALF_OPEN -> CLOSED
```

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

### Interlock Circuit Breaker Quickstart

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

Demonstrates synchronous usage of Interlock CircuitBreaker with decorator, call, and context manager patterns.

```python
from interlock import CircuitBreaker, Config

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

# 1. Decorator — preserves the signature and sync/async nature.
@breaker
def charge(amount: int) -> str:
    return gateway.charge(amount)

# 2. breaker.call — the breaker runs the callable.
result = breaker.call(gateway.charge, 100)

# 3. Context manager — guards a block (exceptions + duration only).
with breaker:
    gateway.charge(100)
```

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

### Install aiohttp integration with pip

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

Install the aiohttp integration for interlock using pip.

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

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

### Installing and Using OTelEventListener

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

Instructions for installing the OpenTelemetry integration and using OTelEventListener to record metrics. This listener records various instruments on the 'interlock' meter.

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

```

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

### Initialize Redis Storage for Circuit Breaker (Sync)

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

Demonstrates how to set up a RedisStorage instance for use with a synchronous CircuitBreaker. Ensure you have the 'redis' library installed.

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

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

### Run Two Clients Example

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

Run the two_clients.py script to see how a Registry manages independent breakers for multiple guarded clients.

```bash
python examples/two_clients.py  # two guarded clients: one fails, the other keeps serving
```

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

### Install Pre-commit Hooks

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

Install the pre-commit git hook for automatically running fast checks.

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

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

### Install Interlock with optional integrations

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

Install Interlock with specific integrations for enhanced functionality, such as FastAPI, Litestar, or Redis.

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

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

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

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

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

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

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

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

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

### Install Interlock with FastAPI

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

Install interlock-cb with optional extras for FastAPI integration, including a 503 Retry-After handler.

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

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

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

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

Use pip to install the interlock-cb[requests] extra for the interlock library.

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

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

### Install interlock with poetry

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

Add the interlock-cb package to your project using poetry.

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

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

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

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

Use pip to install the interlock-cb[httpx2] extra, which includes the httpx2 integration.

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

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

### Full Source Code for two_clients.py

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

This snippet contains the complete source code for the two_clients.py demonstration. It requires the 'interlock' library to be installed.

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

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

### lifecycle.py - Full Source Code

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

The complete source code for the lifecycle.py example, demonstrating a breaker's full state cycle.

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

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

### Install Litestar Extra with pip

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

Use pip to install the interlock-cb[litestar] extra for Litestar version 2.23 and above.

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

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

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

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

Use poetry to add the interlock-cb[httpx2] extra to your project dependencies.

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

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

### Install Interlock with Tenacity Support (uv)

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

Use uv to add the interlock-cb[tenacity] extra, which includes the necessary dependencies for Tenacity integration.

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

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

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

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

Use poetry to add the interlock-cb[requests] extra to your project dependencies.

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

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

### Install Interlock with Tenacity Support (pip)

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

Install the interlock-cb[tenacity] package using pip to enable Tenacity integration.

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

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

### Install Litestar Extra with uv

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

Use uv to add the interlock-cb[litestar] extra for Litestar version 2.23 and above.

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

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

### Custom Listener Implementation Example

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

An example of a custom listener that only implements the on_rejected method to count rejections. Other methods are defined as no-ops.

```python
class RejectionCounter:
    def __init__(self) -> None:
        self.rejected = 0

    def on_rejected(self, *, name: str) -> None:
        self.rejected += 1

    def on_state_change(self, *, name, old, new) -> None: ...
    def on_call(self, *, name, outcome, duration) -> None: ...
    def on_reset(self, *, name) -> None: ...

```

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

### Install Interlock with Tenacity Support (poetry)

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

Add the interlock-cb[tenacity] extra to your project dependencies using poetry.

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

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

### Full Source of pipeline.py

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

This snippet shows the complete source code for the pipeline.py example, illustrating the composition of timeout, circuit breaker, and fallback strategies.

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

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

### Install Litestar Extra with Poetry

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

Use Poetry to add the interlock-cb[litestar] extra for Litestar version 2.23 and above.

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

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

### Add Interlock Redis Dependency with uv

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

Install the Redis extra for Interlock using the uv package manager.

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

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

### Compose Resilience Strategies with Pipeline

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

Use Pipeline.builder() to compose multiple resilience strategies in an explicit order. This example demonstrates fallback, retry, circuit breaker, bulkhead, and timeout.

```python
from interlock import CircuitBreaker, CircuitOpenError, Pipeline

breaker = CircuitBreaker(name='recommendations')

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


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

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

### Install Interlock with Poetry

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

Use Poetry to add the Interlock library with the FastAPI extra to your project dependencies. This command integrates Interlock into your project's dependency management.

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

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

### Clone and Set Up Development Environment

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

Clone the interlock repository and set up the development environment using uv for dependency management.

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

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

### Registry Configuration with Default and Overrides

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

Initialize a Registry with a default configuration and demonstrate per-name configuration overrides.

```python
from interlock import Config, Registry

registry = Registry(config=Config(minimum_number_of_calls=20))

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

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

### Run two_clients.py demo

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

Execute the two_clients.py script to demonstrate handling two clients with one outage and no collateral damage.

```bash
python examples/two_clients.py    # two clients, one outage, no collateral damage
```

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

### Configuring Pipeline with Event Listeners

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

Demonstrates building a pipeline with fallback, circuit breaker, bulkhead, and timeout strategies, attaching a LoggingEventListener to specific components.

```python
from interlock import LoggingEventListener, Pipeline

events = LoggingEventListener()

pipeline = (
    Pipeline.builder()
    .fallback(lambda exc: [], on=(CircuitOpenError,), name='recs', listener=events)
    .circuit_breaker(breaker)  # the breaker keeps its own listener
    .bulkhead(8, name='recs', listener=events)
    .timeout(2.0)
    .build()
)
```

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

### Run lifecycle.py demo

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

Execute the lifecycle.py script to observe a single breaker go through its full state cycle.

```bash
python examples/lifecycle.py      # one breaker through its full state cycle
```

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

### Run pipeline.py demo

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

Execute the pipeline.py script to show a timeout, breaker, and fallback around a hanging service.

```bash
python examples/pipeline.py       # timeout + breaker + fallback around a hanging service
```

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

### Full Source: two_clients.py

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

This script demonstrates two independent circuit breakers managed by a `Registry`. It simulates a partial outage of the 'recommendations' service while the 'payments' service remains available, showing how the application handles failures and recovers.

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

```

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

### Initialize Redis Storage for Circuit Breaker (Async)

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

Shows how to configure an AsyncRedisStorage for use with an asynchronous CircuitBreaker. Requires the 'redis' library with async support.

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

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

### Add Interlock Redis Dependency with pip

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

Install the Redis extra for Interlock using pip.

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

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

### Get CircuitBreaker from Registry

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

Retrieve a named CircuitBreaker from the registry. If the breaker does not exist, it will be created with the provided configuration.

```python
registry.get(name, *, config=None) -> CircuitBreaker
```

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

### Configuring RedisStorage Parameters

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

Instantiate RedisStorage with specific parameters to control key namespace, state lifetime, cache refresh rate, and retry backoff after failures.

```python
RedisStorage(
    client,
    key_prefix='interlock:cb:',  # hash key namespace
    state_ttl=300.0,             # key lifetime (s); refreshed on every write
    poll_interval=1.0,           # cache refresh cadence (s)
    retry_backoff=5.0,           # local-only time after a storage failure (s)
)
```

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

### Basic aiohttp client usage with CircuitBreakerMiddleware

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

Demonstrates how to apply the CircuitBreakerMiddleware to an aiohttp ClientSession to protect requests to a specific host.

```python
import aiohttp

from interlock.integrations.aiohttp import CircuitBreakerMiddleware

middleware = CircuitBreakerMiddleware()

async with aiohttp.ClientSession(middlewares=(middleware,)) as session:
    async with session.get('https://api.example.com/orders') as response:
        orders = await response.json()
```

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

### Registry Methods

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

The Registry creates and caches CircuitBreaker instances. The get method retrieves a CircuitBreaker by name, optionally overriding its configuration during creation.

```python
Registry(*, config=None, clock=None, classifier=None, listener=None, storage=None)
registry.get(name, *, config=None) -> CircuitBreaker
```

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

### Get breaker snapshot

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

Retrieve a snapshot of the breaker's performance metrics, including total calls, failed calls, and failure rate.

```python
breaker.snapshot()
```

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

### fastapi.install_exception_handler

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

Registers a handler mapping CircuitOpenError to 503 with a Retry-After header.

```APIDOC
## install_exception_handler

### Description
Registers an exception handler with a FastAPI application that maps `CircuitOpenError` to an HTTP `503 Service Unavailable` response, including a `Retry-After` header indicating when the circuit may be closed.

### Function Signature
`install_exception_handler(app)`

### Parameters
- `app`: The FastAPI application instance.
```

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

### Python Circuit Breaker Lifecycle Simulation

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

This script simulates the complete lifecycle of a circuit breaker. It requires no special setup beyond running the script.

```python
if __name__ == '__main__':
    main()
```

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

### Create Virtual Environment with uv

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

Use uv to create a new virtual environment for Python 3.12.

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

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

### Basic FastAPI Route with Circuit Breaker Dependency

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

Set up a FastAPI app, install the exception handler, and inject a circuit breaker dependency into a route. This protects the outgoing call to `fetch_orders`.

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