Purgatory

raw JSON →

3.0.1 verified Sat May 09 auth: no python

A circuit breaker implementation for asyncio. Current version 3.0.1, requires Python >=3.9. Releases are sporadic.

pip install purgatory

Common errors

error ModuleNotFoundError: No module named 'purgatory.circuit_breaker' ↓

cause Importing from old submodule path after upgrading to v3.0+.

fix

Use from purgatory import CircuitBreaker instead.

error purgatory.exceptions.CircuitBreakerOpenError ↓

cause Circuit breaker is open; too many failures occurred.

fix

Wait for recovery_timeout seconds before retrying, or implement a retry mechanism with backoff.

Warnings

breaking Version 3.0 changed the import path: CircuitBreaker is now directly importable from `purgatory` instead of `purgatory.circuit_breaker`. ↓

fix Change imports from `from purgatory.circuit_breaker import CircuitBreaker` to `from purgatory import CircuitBreaker`.

gotcha The `CircuitBreaker` context manager does NOT automatically retry. If the breaker is open, it raises `CircuitBreakerOpenError` immediately. You must implement retry logic yourself. ↓

fix Wrap the `async with breaker:` block in a try/except and implement exponential backoff retries.

gotcha The `failure_threshold` counts consecutive failures. A single success resets the counter. This differs from some other circuit breaker implementations (e.g., Hystrix) that use a sliding window. ↓

fix Be aware of the reset behavior; adjust threshold accordingly.

Imports

CircuitBreaker
wrong
```
from purgatory.circuit_breaker import CircuitBreaker
```
correct
```
from purgatory import CircuitBreaker
```
In older versions (pre-3.0) the class was in a submodule. Since 3.0, it's re-exported from the top-level package.

Quickstart

Basic usage of Purgatory's CircuitBreaker context manager.

import asyncio
from purgatory import CircuitBreaker

async def main():
    breaker = CircuitBreaker(
        failure_threshold=5,
        recovery_timeout=30,
        half_open_max_calls=1
    )

    for _ in range(10):
        async with breaker:
            print("Call succeeded")

asyncio.run(main())