semvec/pypi-packages huggingface dataset:
ss-utils-circuit-breaker 0.0.1
Async circuit breaker with optional Redis (coredis) support for distributed state
License: MIT
Requires Python: >=3.10
Keywords: async, circuit-breaker, coredis, redis, resilience

# ss-utils-circuit-breaker

A Python async circuit breaker implementation with optional Redis (coredis) support for distributed state.

## Installation

```bash
# Basic installation (in-memory state only)
uv add ss-utils-circuit-breaker

# With Redis support
uv add ss-utils-circuit-breaker[redis]
```

## Quick Start

### Basic Usage (In-Memory)

```python
from ss_utils_circuit_breaker import CircuitBreaker, CircuitBreakerError

# Create a circuit breaker
breaker = CircuitBreaker(
    name="my_api",
    failure_threshold=5,  # Open after 5 failures
    recovery_timeout=60,  # Try again after 60 seconds
)

@breaker
async def call_external_api():
    # Your async API call here
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        response.raise_for_status()
        return response.json()

# Usage
try:
    data = await call_external_api()
except CircuitBreakerError:
    # Circuit is open - service is known to be unavailable
    data = get_cached_data()
except Exception:
    # Regular failure
    pass
```

### With Redis (Distributed State)

```python
from ss_utils_circuit_breaker import RedisCircuitBreaker

# Create a Redis-backed circuit breaker
breaker = RedisCircuitBreaker(
    name="my_api",
    redis_url="localhost:6379",
    namespace="circuit_breaker",  # Redis key prefix
    failure_threshold=5,
    recovery_timeout=60,
)

@breaker
async def call_external_api():
    # State is shared across all instances via Redis
    pass
```

## Circuit Breaker States

```
CLOSED (Normal)
    │
    ▼ (failures >= threshold)
   OPEN (Blocking)
    │
    ▼ (after recovery_timeout)
HALF_OPEN (Testing)
    │
    ├─► (success) ──► CLOSED
    │
    └─► (failure) ──► OPEN
```

- **CLOSED**: Normal operation. Calls pass through, failures are counted.
- **OPEN**: Circuit is tripped. Calls fail immediately with `CircuitBreakerError`.
- **HALF_OPEN**: After timeout, one test call is allowed. Success closes circuit, failure reopens it.



## Configuration Example

```python
import httpx
from ss_utils_circuit_breaker import CircuitBreaker

# Only count connection errors as failures
api_breaker = CircuitBreaker(
    name="external_api",
    failure_threshold=3,
    recovery_timeout=30,
    expected_exception=(httpx.ConnectError, httpx.TimeoutException),
)

@api_breaker
async def fetch_data():
    async with httpx.AsyncClient(timeout=10.0) as client:
        return await client.get("https://api.example.com/data")
```

## Testing

```bash
# Run tests
uv run pytest

# With coverage
uv run pytest --cov
```

## License

MIT