### Complete Purgatory Configuration Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Demonstrates a full setup including Redis-backed unit of work, custom default circuit breaker configurations, and adding event listeners. Shows how to use different configurations for individual circuit breakers.

```python
import asyncio
from purgatory import (
    AsyncCircuitBreakerFactory,
    AsyncRedisUnitOfWork,
)

async def main():
    # Create unit of work with Redis backend
    uow = AsyncRedisUnitOfWork("redis://localhost:6379")
    await uow.initialize()
    
    # Create factory with custom defaults
    factory = AsyncCircuitBreakerFactory(
        default_threshold=5,
        default_ttl=60.0,
        exclude=[ValueError],  # Exclude client errors globally
        uow=uow
    )
    await factory.initialize()
    
    # Add monitoring
    def log_events(name: str, event_type: str, event: Event) -> None:
        print(f"[{event_type:20}] {name:15} — {event}")
    
    factory.add_listener(log_events)
    
    # Use different configurations per circuit
    
    # Fast API - tolerant
    async with await factory.get_breaker("fast_api", threshold=10) as breaker:
        result = await api_call()
    
    # Slow API - sensitive, long recovery
    async with await factory.get_breaker(
        "slow_api",
        threshold=3,
        ttl=120,
        exclude=[TimeoutError]
    ) as breaker:
        result = await slow_operation()

asyncio.run(main())
```

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

### AsyncRedisUnitOfWork Usage Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Demonstrates creating an AsyncRedisUnitOfWork, initializing it, and then using it with an AsyncCircuitBreakerFactory. Ensure the Redis package is installed and the URL is correct.

```python
from purgatory import AsyncRedisUnitOfWork, AsyncCircuitBreakerFactory

uow = AsyncRedisUnitOfWork("redis://localhost:6379")
await uow.initialize()

factory = AsyncCircuitBreakerFactory(uow=uow)
await factory.initialize()
```

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

### SyncRedisUnitOfWork Usage Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Instantiates SyncRedisUnitOfWork, initializes it, and then creates a SyncCircuitBreakerFactory. Ensure the Redis package is installed.

```python
from purgatory import SyncRedisUnitOfWork, SyncCircuitBreakerFactory

uow = SyncRedisUnitOfWork("redis://localhost:6379")
uow.initialize()

factory = SyncCircuitBreakerFactory(uow=uow)
factory.initialize()
```

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

### Install Purgatory

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Install the Purgatory library using pip. For Redis support, include the optional redis extra.

```bash
pip install purgatory
```

```bash
pip install purgatory[redis]
```

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

### Install Redis Package for Purgatory

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Installs the Purgatory package with Redis support. Alternatively, install the redis library separately.

```bash
pip install purgatory[redis]
```

```bash
pip install redis>=5.2.0,<6
```

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

### AsyncRedisRepository Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Initializes the Redis-backed repository. Requires a Redis connection URL. Throws ConfigurationError if the redis package is not installed.

```python
class AsyncRedisRepository(AsyncAbstractRepository):
    def __init__(self, url: str) -> None
```

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

### Initialize Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

The initialize method is used to set up repositories and connections. Subclasses should override this for specific setup needs. The default implementation does nothing.

```python
def initialize(self) -> None
```

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

### Check Redis Installation for Distributed Setup

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/errors.md

When setting up a distributed circuit breaker, check if Redis is available. If not, fall back to using an in-memory storage solution.

```python
try:
    from purgatory import AsyncRedisUnitOfWork
    uow = AsyncRedisUnitOfWork("redis://localhost:6379")
except ConfigurationError:
    print("Redis not available, using in-memory storage")
    from purgatory import AsyncInMemoryUnitOfWork
    uow = AsyncInMemoryUnitOfWork()
```

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

### Handle Redis Not Installed Error

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/errors.md

Demonstrates how to catch a ConfigurationError when Redis is not installed or accessible, and includes a fallback to in-memory storage.

```python
from purgatory import AsyncRedisUnitOfWork, AsyncCircuitBreakerFactory

try:
    uow = AsyncRedisUnitOfWork("redis://localhost:6379")  # Raises ConfigurationError
except ConfigurationError as e:
    print(f"Configuration error: {e}")
    # Fallback to in-memory storage
```

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

### HalfOpenedState Example Usage

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/States.md

Demonstrates how the HalfOpenedState behaves with a test request. Shows transitions based on success or failure of the test request.

```python
context = Context("api", threshold=3, ttl=30)
context.set_state(HalfOpenedState())
# State is HalfOpenedState, failure_count = None

# Test request succeeds
# → handle_end_request() called
# → Emit RECOVERED event
# → Transition to CLOSED with failure_count=0

# OR: Test request fails
# → handle_exception() called
# → Transition to OPENED
```

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

### SyncInMemoryUnitOfWork Usage Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Instantiates SyncInMemoryUnitOfWork and SyncCircuitBreakerFactory. This example demonstrates basic usage with a circuit breaker.

```python
from purgatory import SyncInMemoryUnitOfWork, SyncCircuitBreakerFactory

uow = SyncInMemoryUnitOfWork()
factory = SyncCircuitBreakerFactory(uow=uow)

with factory.get_breaker("my_circuit") as breaker:
    result = api_call()
```

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

### AsyncInMemoryUnitOfWork Usage Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Demonstrates creating an AsyncInMemoryUnitOfWork and using it with an AsyncCircuitBreakerFactory. No explicit initialization is needed for the in-memory version.

```python
from purgatory import AsyncInMemoryUnitOfWork, AsyncCircuitBreakerFactory

uow = AsyncInMemoryUnitOfWork()
factory = AsyncCircuitBreakerFactory(uow=uow)

async with await factory.get_breaker("my_circuit") as breaker:
    result = await api_call()
```

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

### Example Usage of Hook

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/types.md

Demonstrates adding a listener function to an AsyncCircuitBreakerFactory to handle circuit breaker events.

```python
def handle_circuit_events(
    name: str,
    event_type: Literal["circuit_breaker_created", "state_changed", "failed", "recovered"],
    event: Event
) -> None:
    if event_type == "state_changed":
        print(f"Circuit {name} changed state")
    elif event_type == "failed":
        print(f"Circuit {name} has a failure")

factory = AsyncCircuitBreakerFactory()
factory.add_listener(handle_circuit_events)
```

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

### Install Redis Dependency for Purgatory

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/storage_backend.md

Install the 'redis' extra dependency for Purgatory to enable Redis support. This is required for both asynchronous and synchronous configurations.

```bash
pip install "purgatory[redis]"
```

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

### Async In-Memory UnitOfWork and CircuitBreakerFactory

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Demonstrates the setup for asynchronous in-memory circuit breakers.

```python
# Async
from purgatory import AsyncInMemoryUnitOfWork, AsyncCircuitBreakerFactory

uow = AsyncInMemoryUnitOfWork()
factory = AsyncCircuitBreakerFactory(uow=uow)
```

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

### Redis URL Format Examples

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Illustrates various formats for specifying Redis connection URLs, including authentication and database selection.

```plaintext
redis://[:password@]host:port[/db]
```

```plaintext
redis://localhost:6379
```

```plaintext
redis://localhost:6379/0
```

```plaintext
redis://:password@localhost:6379
```

```plaintext
redis://redis.example.com:6379
```

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

### Sync Redis UnitOfWork and CircuitBreakerFactory Initialization

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Illustrates the initialization process for synchronous Redis-backed circuit breakers, including Redis connection and factory setup.

```python
# Sync
from purgatory import SyncRedisUnitOfWork, SyncCircuitBreakerFactory

uow = SyncRedisUnitOfWork("redis://localhost:6379")
uow.initialize()
factory = SyncCircuitBreakerFactory(uow=uow)
factory.initialize()
```

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

### Sync In-Memory UnitOfWork and CircuitBreakerFactory

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Demonstrates the setup for synchronous in-memory circuit breakers.

```python
# Sync
from purgatory import SyncInMemoryUnitOfWork, SyncCircuitBreakerFactory

uow = SyncInMemoryUnitOfWork()
factory = SyncCircuitBreakerFactory(uow=uow)
```

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

### ClosedState Behavior Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/States.md

Illustrates the state transitions of ClosedState based on request outcomes. Shows how failure counts increment and trigger a transition to the OPENED state.

```python
context = Context("api", threshold=3, ttl=30)
# State is ClosedState with failure_count=0

# Request 1: Success
# → handle_end_request() called
# → failure_count=0, stays CLOSED

# Request 2: Fails
# → handle_exception() called
# → failure_count=1, stays CLOSED

# Request 3: Fails
# → handle_exception() called
# → failure_count=2, stays CLOSED

# Request 4: Fails
# → handle_exception() called
# → failure_count=3 >= threshold
# → Transition to OPENED
```

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

### Circuit Breaker Event Flow and Logging

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Events.md

Demonstrates how to set up logging for circuit breaker events and illustrates the event flow during normal operation, failures, and state changes. This example requires the AsyncCircuitBreakerFactory and Event classes.

```python
from purgatory import AsyncCircuitBreakerFactory, Event

# Set up logging
def log_events(name: str, event_type: str, event: Event) -> None:
    if isinstance(event, CircuitBreakerCreated):
        print(f"[CREATED] {event.name}: threshold={event.threshold}, ttl={event.ttl}")
    elif isinstance(event, ContextChanged):
        print(f"[STATE] {event.name} ">→</ {event.state}")
    elif isinstance(event, CircuitBreakerFailed):
        print(f"[FAILED] {event.name}: count={event.failure_count}")
    elif isinstance(event, CircuitBreakerRecovered):
        print(f"[RECOVERED] {event.name}")

factory = AsyncCircuitBreakerFactory(default_threshold=2)
factory.add_listener(log_events)

# Usage generates events:
async with await factory.get_breaker("api") as breaker:
    await risky_operation()

# Output:
# [CREATED] api: threshold=2, ttl=30
# (first success)
# [RECOVERED] api

# On second call with failure:
try:
    async with await factory.get_breaker("api") as breaker:
        raise Exception("API error")
except:
    pass

# Output:
# [FAILED] api: count=1

# On third call with second failure:
try:
    async with await factory.get_breaker("api") as breaker:
        raise Exception("API error")
except:
    pass

# Output:
# [FAILED] api: count=2
# [STATE] api ">→</ opened
```

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

### Configure Asynchronous Redis Storage Backend

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/storage_backend.md

Sets up an asynchronous circuit breaker factory using Redis as the unit of work. Ensure the 'redis' extra is installed and the 'initialize' coroutine is called.

```python
import httpx
from purgatory import AsyncCircuitBreakerFactory, AsyncRedisUnitOfWork

circuit_breaker = AsyncCircuitBreakerFactory(
   default_threshold=5,
   default_ttl=30,
   exclude=[
      (httpx.HTTPStatusError, lambda exc: exc.response.is_client_error),
   ],
   uow=AsyncRedisUnitOfWork("redis://localhost/0"),
)

await circuit_breaker.initialize()
```

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

### Initialize AsyncCircuitBreakerFactory with Redis

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Initializes the factory, which is necessary if using persistent storage like Redis. Ensure the 'redis' package is installed if Redis is configured.

```python
from purgatory import AsyncCircuitBreakerFactory, AsyncRedisUnitOfWork

factory = AsyncCircuitBreakerFactory(
    uow=AsyncRedisUnitOfWork("redis://localhost:6379")
)
await factory.initialize()
```

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

### SyncCircuitBreaker __enter__ Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreaker.md

Demonstrates how to use the SyncCircuitBreaker as a context manager. The 'with' statement ensures that circuit breaker logic is applied before and after the enclosed code block.

```python
factory = SyncCircuitBreakerFactory()
breaker = factory.get_breaker("my_circuit")

with breaker:
    result = api_call()
```

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

### Example Usage of ExcludeExcType

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/types.md

Demonstrates creating an AsyncCircuitBreakerFactory excluding a specific exception type.

```python
factory = AsyncCircuitBreakerFactory(exclude=[ValueError])
```

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

### Initialize SyncCircuitBreakerFactory with Redis

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreakerFactory.md

Initializes the factory and its repositories. This is required before using the factory if persistent storage like Redis is configured. Ensure the 'redis' package is installed if Redis is used.

```python
from purgatory import SyncCircuitBreakerFactory, SyncRedisUnitOfWork

factory = SyncCircuitBreakerFactory(
    uow=SyncRedisUnitOfWork("redis://localhost:6379")
)
factory.initialize()
```

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

### OpenedState Example Usage

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/States.md

Demonstrates the behavior of the OpenedState based on time relative to its TTL. It shows the state transition from OPENED to HALF-OPENED after TTL expiration and re-processing of a request.

```python
# Assuming Context and OpenedState are imported and initialized
# context = Context("api", threshold=1, ttl=5)
# context.set_state(OpenedState("api"))  # Time = 1000.0, opened_at = 1000.0

# Immediately after opening
# current_time = 1000.5
# closed_at = 1000.0 + 5 = 1005.0
# 1000.5 > 1005.0? NO
# → Raise OpenedState

# After TTL expires
# current_time = 1006.0
# closed_at = 1000.0 + 5 = 1005.0
# 1006.0 > 1005.0? YES
# → Transition to HALF-OPENED
# → Re-process request in HALF-OPENED state
```

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

### Configure Synchronous Redis Storage Backend

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/storage_backend.md

Sets up a synchronous circuit breaker factory using Redis as the unit of work. Ensure the 'redis' extra is installed. The 'initialize' method is autogenerated but cannot be used for synchronous clients.

```python
import requests
from purgatory import SyncCircuitBreakerFactory, SyncRedisUnitOfWork

circuit_breaker = SyncCircuitBreakerFactory(
   default_threshold=5,
   default_ttl=30,
   exclude=[
      (requests.HTTPError, lambda exc: 400 <= exc.response.status_code < 500),
   ],
   uow=SyncRedisUnitOfWork("redis://localhost/0"),
)
```

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

### Redis Key Structure Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Shows the key naming convention used by Purgatory in Redis for storing circuit breaker state and failure counts.

```plaintext
cbr::{circuit_name}                — Main context (JSON)
cbr::{circuit_name}::failure_count — Failure counter
```

```plaintext
cbr::api          → {"name":"api","threshold":5,"ttl":30,"state":"closed","opened_at":null}
cbr::api::failure_count → 0
```

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

### Example Usage of ExcludeType

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/types.md

Illustrates configuring an AsyncCircuitBreakerFactory with a list of exceptions to exclude, including a custom predicate.

```python
exclude = [
    AuthenticationError,  # Exclude this type completely
    (ValueError, lambda e: "retry" in str(e))  # Exclude ValueError with predicate
]

factory = AsyncCircuitBreakerFactory(exclude=exclude)
```

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

### SyncCircuitBreaker __exit__ Example with Exception

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreaker.md

Shows how the SyncCircuitBreaker handles exceptions within its context. An exception is automatically counted as a failure, and the circuit breaker is notified.

```python
# Exception automatically counted as failure
try:
    with breaker:
        raise ValueError("API error")
except ValueError:
    print("Exception was raised but circuit breaker was notified")
```

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

### AsyncCircuitBreaker Usage Example

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreaker.md

Demonstrates how to obtain and use an AsyncCircuitBreaker instance within an async with block to protect an API call. Ensure AsyncCircuitBreakerFactory is initialized.

```python
factory = AsyncCircuitBreakerFactory()
breaker = await factory.get_breaker("my_circuit")

async with breaker:
    result = await api_call()
```

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

### Async Circuit Breaker Initialization and Usage

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/README.md

Initializes an AsyncCircuitBreakerFactory and uses a breaker for an asynchronous API call. Ensure 'purgatory' is installed and 'external_api_call' is defined.

```python
from purgatory import AsyncCircuitBreakerFactory

factory = AsyncCircuitBreakerFactory(default_threshold=5, default_ttl=30)

async with await factory.get_breaker("api") as breaker:
    result = await external_api_call()
```

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

### Sync Circuit Breaker Initialization and Usage

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/README.md

Initializes a SyncCircuitBreakerFactory and uses a breaker for a synchronous API call. Ensure 'purgatory' is installed and 'external_api_call' is defined.

```python
from purgatory import SyncCircuitBreakerFactory

factory = SyncCircuitBreakerFactory(default_threshold=5, default_ttl=30)

with factory.get_breaker("api") as breaker:
    result = external_api_call()
```

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

### SyncRedisRepository Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Defines the constructor for the synchronous Redis-backed repository. It requires a Redis connection URL and initializes attributes like 'redis', 'messages', and 'prefix'. Throws ConfigurationError if the redis package is not installed.

```python
class SyncRedisRepository(SyncAbstractRepository):
    def __init__(self, url: str) -> None:
```

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

### Example Usage of CreateCircuitBreaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Commands.md

Demonstrates how a circuit breaker is created internally when requesting a non-existent breaker via the factory. The factory handles the creation and processing of the CreateCircuitBreaker command.

```python
# Internal: Automatically triggered when getting a non-existent breaker
factory = AsyncCircuitBreakerFactory()
breaker = await factory.get_breaker("new_circuit", threshold=5, ttl=30)

# The factory internally creates and processes:
# cmd = CreateCircuitBreaker("new_circuit", 5, 30)
# context = await messagebus.handle(cmd, uow)
```

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

### SyncCircuitBreaker __exit__ Example with Success

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreaker.md

Illustrates a successful execution within the SyncCircuitBreaker context. A successful operation resets the failure counter and can transition the circuit state back to CLOSED if it was in HALF-OPENED.

```python
# Success resets failure counter
with breaker:
    result = api_call()  # If succeeds, failure count resets
```

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

### Register Listener for ContextChanged Event

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Events.md

Example of registering a listener to handle ContextChanged events. This allows monitoring circuit breaker state transitions and reacting to changes, such as logging the new state of a circuit.

```python
def on_state_changed(name: str, event_type: str, event: Event) -> None:
    if isinstance(event, ContextChanged):
        print(f"Circuit '{event.name}' is now {event.state}")

factory.add_listener(on_state_changed)
```

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

### Distributed Circuit Breaker Setup with Redis

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Configure Purgatory to use a distributed backend, such as Redis, for state management. This allows multiple processes to share the same circuit breaker state, ensuring consistent behavior across a distributed system.

```python
# All processes share state via Redis
uow = AsyncRedisUnitOfWork("redis://redis-server:6379")
await uow.initialize()

factory = AsyncCircuitBreakerFactory(uow=uow)
await factory.initialize()
```

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

### Registering a Listener for CircuitBreakerFailed Events

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Events.md

Example of how to register a listener function to handle CircuitBreakerFailed events. The listener function checks the event type and prints the circuit name and failure count if it's a CircuitBreakerFailed event. This demonstrates how to hook into circuit breaker failures.

```python
def on_failure(name: str, event_type: str, event: Event) -> None:
    if isinstance(event, CircuitBreakerFailed):
        print(f"Circuit '{event.name}' failed: count={event.failure_count}")

factory.add_listener(on_failure)
```

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

### Complete Exception Handling with Async Circuit Breaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/errors.md

An example showcasing robust exception handling for circuit breaker operations, including catching specific exceptions like OpenedState and ConnectionError, as well as general exceptions. This pattern is useful for managing unreliable services.

```python
import asyncio
from purgatory import AsyncCircuitBreakerFactory, Event
from purgatory.domain.model import OpenedState

async def main():
    factory = AsyncCircuitBreakerFactory(
        default_threshold=3,
        default_ttl=5
    )
    
    # Track events for debugging
    def log_events(name: str, event_type: str, event: Event) -> None:
        print(f"[{event_type}] {name}: {event}")
    
    factory.add_listener(log_events)
    
    # Simulate failing requests
    async def unreliable_api():
        import random
        if random.random() < 0.8:  # 80% failure rate
            raise ConnectionError("Service unavailable")
        return {"status": "ok"}
    
    # Call with circuit breaker protection
    breaker = await factory.get_breaker("api")
    
    for attempt in range(10):
        try:
            async with breaker:
                result = await unreliable_api()
                print(f"Attempt {attempt}: Success - {result}")
        except OpenedState:
            print(f"Attempt {attempt}: Circuit is open, skipping")
        except ConnectionError as e:
            print(f"Attempt {attempt}: Request failed - {e}")
        except Exception as e:
            print(f"Attempt {attempt}: Unexpected error - {e}")
        
        await asyncio.sleep(1)

asyncio.run(main())
```

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

### Registering a Listener for CircuitBreakerRecovered Events

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Events.md

Example of registering a listener to handle CircuitBreakerRecovered events. The listener function identifies CircuitBreakerRecovered events and prints a message indicating the circuit has recovered. This shows how to be notified of successful circuit breaker recovery.

```python
def on_recovery(name: str, event_type: str, event: Event) -> None:
    if isinstance(event, CircuitBreakerRecovered):
        print(f"Circuit '{event.name}' has recovered")

factory.add_listener(on_recovery)
```

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

### initialize

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Initializes the repository and its connections. This method is intended to be overridden by concrete implementations.

```APIDOC
## initialize

### Description
Initialize the repository and its connections. Default does nothing; override in subclasses.

### Method
`async def initialize(self) -> None`

### Parameters
None

### Return
None

### Throws
Connection errors
```

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

### AsyncRedisRepository Initialize Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Connects to Redis. This method must be called before using the repository. It can throw connection errors.

```python
async def initialize(self) -> None
```

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

### initialize

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Initializes the unit of work and repositories. This should be called before using the factory if persistent storage (Redis) is being used.

```APIDOC
## initialize

### Description
Initialize the unit of work and repositories. Call this before using the factory if using persistent storage (Redis).

### Method
`async def initialize(self) -> None`

### Parameters
None

### Return
None

### Throws
`ConfigurationError` if Redis is configured but the redis package is not installed

### Example
```python
from purgatory import AsyncCircuitBreakerFactory, AsyncRedisUnitOfWork

factory = AsyncCircuitBreakerFactory(
    uow=AsyncRedisUnitOfWork("redis://localhost:6379")
)
await factory.initialize()
```
```

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

### Initialize SyncCircuitBreakerFactory with Redis Backend

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Sets up a SyncCircuitBreakerFactory using a Redis backend for storage. Ensure Redis is running and accessible.

```python
from purgatory import SyncCircuitBreakerFactory, SyncRedisUnitOfWork

uow = SyncRedisUnitOfWork("redis://localhost:6379")
uow.initialize()
factory = SyncCircuitBreakerFactory(uow=uow)
factory.initialize()
```

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

### AsyncAbstractRepository Initialize Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Initializes the repository and its connections. This method is intended to be overridden in subclasses.

```python
async def initialize(self) -> None:
```

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

### Async Redis UnitOfWork and CircuitBreakerFactory Initialization

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Shows how to initialize asynchronous Redis-backed circuit breakers, including connecting to Redis and initializing the factory.

```python
# Async
from purgatory import AsyncRedisUnitOfWork, AsyncCircuitBreakerFactory

uow = AsyncRedisUnitOfWork("redis://localhost:6379")
await uow.initialize()
factory = AsyncCircuitBreakerFactory(uow=uow)
await factory.initialize()
```

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

### Example Usage of ExcludeTypeFunc

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/types.md

Demonstrates creating an AsyncCircuitBreakerFactory with a custom predicate function for excluding specific exceptions.

```python
def is_retryable(exc: BaseException) -> bool:
    return "retry" in str(exc).lower()

factory = AsyncCircuitBreakerFactory(
    exclude=[(ValueError, is_retryable)]
)
```

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

### get

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Retrieves a circuit breaker context by its name. Returns the context if found, otherwise returns None.

```APIDOC
## get

### Description
Retrieve a circuit breaker context by name.

### Method
`async def get(self, name: CircuitName) -> Optional[Context]`

### Parameters
#### Path Parameters
- **name** (str) - Yes - Circuit name

### Return
`Optional[Context]` — Circuit context or None if not found

### Throws
None
```

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

### Initialize SyncCircuitBreakerFactory with Default Configuration

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Instantiates a SyncCircuitBreakerFactory using default settings for threshold, TTL, and in-memory storage.

```python
from purgatory import SyncCircuitBreakerFactory

factory = SyncCircuitBreakerFactory()
# Uses: threshold=5, ttl=30, in-memory storage
```

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

### AsyncAbstractRepository Get Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Retrieves a circuit breaker context by its name. Returns None if the circuit is not found.

```python
async def get(self, name: CircuitName) -> Optional[Context]:
```

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

### get_breaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreakerFactory.md

Gets or creates a sync circuit breaker. Returns the same breaker instance for the same circuit name on subsequent calls.

```APIDOC
## get_breaker

### Description
Get or create a sync circuit breaker. Returns the same breaker instance for the same circuit name on subsequent calls.

### Method
`get_breaker(circuit: CircuitName, threshold: Optional[Threshold] = None, ttl: Optional[TTL] = None, exclude: Optional[ExcludeType] = None)`

### Parameters
#### Path Parameters
- **circuit** (`str`) - Required - Unique circuit identifier
- **threshold** (`Optional[int]`) - Optional - Override default threshold for this breaker
- **ttl** (`Optional[float]`) - Optional - Override default TTL for this breaker
- **exclude** (`Optional[ExcludeType]`) - Optional - Override exception exclusion list for this breaker

### Return
`SyncCircuitBreaker` - Context manager for executing protected code

### Throws
`ConfigurationError` if Redis is unavailable and configured

### Example
```python
factory = SyncCircuitBreakerFactory(default_threshold=3, default_ttl=60)

# Get a breaker and use as context manager
with factory.get_breaker("api_call") as breaker:
    result = external_api_call()

# Override defaults per breaker
with factory.get_breaker(
    "slow_api",
    threshold=10,
    ttl=120
) as breaker:
    result = slow_operation()
```
```

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

### get_breaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Gets or creates an asynchronous circuit breaker. Returns the same breaker instance for the same circuit name on subsequent calls.

```APIDOC
## get_breaker

### Description
Get or create an async circuit breaker. Returns the same breaker instance for the same circuit name on subsequent calls.

### Method
`async def get_breaker(self, circuit: CircuitName, threshold: Optional[Threshold] = None, ttl: Optional[TTL] = None, exclude: Optional[ExcludeType] = None) -> AsyncCircuitBreaker`

### Parameters
#### Path Parameters
- **circuit** (`str`) - Required - Unique circuit identifier
- **threshold** (`Optional[int]`) - Optional - Override default threshold for this breaker
- **ttl** (`Optional[float]`) - Optional - Override default TTL for this breaker
- **exclude** (`Optional[ExcludeType]`) - Optional - Override exception exclusion list for this breaker

### Return
`AsyncCircuitBreaker` - Context manager for executing protected code

### Throws
`ConfigurationError` if Redis is unavailable and configured

### Example
```python
factory = AsyncCircuitBreakerFactory(default_threshold=3, default_ttl=60)

# Get a breaker and use as context manager
async with await factory.get_breaker("api_call") as breaker:
    result = await external_api_call()

# Override defaults per breaker
async with await factory.get_breaker(
    "slow_api",
    threshold=10,
    ttl=120
) as breaker:
    result = await slow_operation()
```
```

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

### Get Current Circuit State

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Context.md

Retrieves the current state of the circuit breaker. The state can be 'closed', 'opened', or 'half-opened'.

```python
@property
def state(self) -> StateName:
    # Current circuit state: "closed", "opened", or "half-opened"
    pass
```

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

### Exclude Specific Exceptions

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Use the 'exclude' parameter to prevent certain exceptions from being counted as failures. This example excludes AuthenticationError.

```python
async with await factory.get_breaker(
    "api",
    exclude=[AuthenticationError]
) as breaker:
    await api_call()
```

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

### Get Current Failure Count

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Context.md

Retrieves the current number of failures recorded by the circuit breaker. This count is used for state tracking.

```python
@property
def failure_count(self) -> Optional[int]:
    # Current failure count for state tracking.
    pass
```

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

### Exclude Specific Exceptions

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreakerFactory.md

Use the 'exclude' parameter to prevent certain exceptions from triggering the circuit breaker. This example shows how to exclude 'AuthenticationError'.

```python
from purgatory import AuthenticationError

# Exclude authentication errors
with factory.get_breaker(
    "api",
    exclude=[AuthenticationError]
) as breaker:
    api_call()
```

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

### AsyncRedisRepository.initialize

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Connects to Redis. This method must be called before using other repository methods.

```APIDOC
## AsyncRedisRepository.initialize

### Description
Connects to Redis. Must be called before use.

### Method
`async def initialize(self) -> None`

### Parameters
None

### Return
None

### Throws
Connection errors
```

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

### AsyncRedisUnitOfWork.initialize

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Connects to the Redis instance. This method must be called before using the AsyncRedisUnitOfWork.

```APIDOC
## AsyncRedisUnitOfWork.initialize

### Description
Connects to the Redis instance. This method must be called before using the AsyncRedisUnitOfWork.

### Method
```python
async def initialize(self) -> None
```

### Parameters
None

### Return
None

### Throws
- Connection errors from Redis
```

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

### Configuring and Using AsyncCircuitBreaker in a Usage Pattern

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreaker.md

Illustrates a common usage pattern for AsyncCircuitBreaker, including configuration of the factory with default settings and obtaining a breaker with specific parameters for an API call. It also shows how to catch the OpenedState exception.

```python
factory = AsyncCircuitBreakerFactory(default_threshold=3)

async def call_external_api():
    breaker = await factory.get_breaker("external_api", threshold=5, ttl=60)
    try:
        async with breaker:
            response = await httpx.get("https://api.example.com/data")
            return response.json()
    except OpenedState as e:
        logger.warning(f"Circuit breaker is open: {e}")
        return None
```

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

### Get Circuit Opened Timestamp

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Context.md

Retrieves the Unix timestamp when the circuit breaker transitioned to the OPENED state. Returns None if the circuit has never been opened.

```python
@property
def opened_at(self) -> Optional[float]:
    # Unix timestamp when circuit transitioned to OPENED state, or None if never opened.
    pass
```

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

### Custom Command Definition and Registration

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Commands.md

Shows how to define and register a custom command by extending the base Command class. This includes defining the command structure and registering its handler with the message registry.

```python
from dataclasses import dataclass
from purgatory.domain.messages.base import Command
from purgatory.typing import CircuitName

@dataclass(frozen=True)
class CustomCommand(Command):
    circuit_name: CircuitName
    custom_data: str

# Register handler
async def handle_custom_command(cmd: CustomCommand, uow) -> Any:
    # Implement custom logic
    pass

# Use in factory
registry = AsyncMessageRegistry()
registry.add_listener(CustomCommand, handle_custom_command)
```

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

### AsyncInMemoryRepository Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Initializes the in-memory repository. State changes are stored directly in Context objects.

```python
class AsyncInMemoryRepository(AsyncAbstractRepository):
    def __init__(self) -> None
```

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

### Initialize AsyncCircuitBreakerFactory with Redis Persistence

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Configure the factory to use Redis for distributed persistence. Ensure Redis is running and accessible.

```python
from purgatory import AsyncRedisUnitOfWork

uow = AsyncRedisUnitOfWork("redis://localhost:6379")
await uow.initialize()
factory = AsyncCircuitBreakerFactory(uow=uow)
```

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

### AsyncCircuitBreakerFactory with Redis Backend

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Sets up the circuit breaker factory to use Redis for state storage. Requires initializing the factory after instantiation.

```python
from purgatory import AsyncCircuitBreakerFactory, AsyncRedisUnitOfWork

uow = AsyncRedisUnitOfWork("redis://localhost:6379")
factory = AsyncCircuitBreakerFactory(uow=uow)
await factory.initialize()  # Must call before using
```

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

### Get Circuit Breaker with Per-Breaker Overrides

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Obtain a circuit breaker instance, overriding factory default settings for threshold, TTL, and excluded exceptions for a specific circuit.

```python
async with await factory.get_breaker(
    "my_circuit",
    threshold=10,             # Override default
    ttl=60,                   # Override default
    exclude=[TimeoutError],   # Override exclusions
) as breaker:
    result = await operation()
```

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

### Default AsyncCircuitBreakerFactory Configuration

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Instantiates the factory with default in-memory storage, threshold of 5, and TTL of 30 seconds.

```python
from purgatory import AsyncCircuitBreakerFactory

factory = AsyncCircuitBreakerFactory()
# Uses: threshold=5, ttl=30, in-memory storage
```

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

### SyncCircuitBreaker Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreaker.md

Illustrates the parameters for the SyncCircuitBreaker constructor. Note that direct instantiation is not recommended; use SyncCircuitBreakerFactory.get_breaker() instead.

```python
SyncCircuitBreaker(
    context: Context,
    uow: SyncAbstractUnitOfWork,
    messagebus: SyncMessageRegistry,
)
```

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

### Initialize AsyncCircuitBreakerFactory

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/introduction.md

Create an instance of AsyncCircuitBreakerFactory with default failure threshold, TTL, and a list of exceptions to exclude from failure tracking.

```python
import httpx
from purgatory import AsyncCircuitBreakerFactory

circuitbreaker = AsyncCircuitBreakerFactory(
   default_threshold=5,
   default_ttl=30,
   exclude=[
      (httpx.HTTPStatusError, lambda exc: exc.response.is_client_error),
   ]
)
```

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

### AsyncRedisRepository Get Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Repository.md

Loads a circuit breaker's context from Redis. Reconstructs the Context from stored JSON. It loads both circuit state and failure count from separate keys.

```python
async def get(self, name: CircuitName) -> Optional[Context]
```

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

### Prometheus Hook for Circuit Breaker Metrics

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/monitoring.md

Implement a Prometheus hook to collect state changes and errors from circuit breakers. Register this hook with the AsyncCircuitBreakerFactory to start collecting metrics.

```python
from prometheus_client import Counter, Gauge
from purgatory import AsyncCircuitBreakerFactory


class GaugeStateValue:
   CLOSED = 0
   HALF_OPEN = 1
   OPEN = 2


class PrometheusHook:
   def __init__(self):
      circuit_breaker_error = Counter(
         "circuit_breaker_error",
         "Count the circuit breaker exception raised",
         labelnames=["circuit"],
      )

      circuit_breaker_state = Gauge(
         "circuit_breaker_state",
         "State of the circuit breaker. 0 is closed, 1 is half-opened, 2 is opened.",
         labelnames=["circuit"],
      )

   def __call__(self, circuit_name: str, evt_type: str, payload: Any) -> None:
      if evt_type == "state_changed":
            state = {
               "closed": GaugeStateValue.CLOSED,
               "half-opened": GaugeStateValue.HALF_OPEN,
               "opened": GaugeStateValue.OPEN,
            }[payload.state]
            self.prometheus_metrics.blacksmith_circuit_breaker_state.labels(
               circuit_name
            ).set(state)
      elif evt_type == "failed":
            self.prometheus_metrics.blacksmith_circuit_breaker_error.labels(
               circuit_name
            ).inc()


circuitbreaker = AsyncCircuitBreakerFactory()
circuitbreaker.add_listener(PrometheusHook())
```

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

### AsyncAbstractUnitOfWork Initialize Method

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

The 'initialize' method is an asynchronous function intended for subclasses to override for setting up repositories and connections. The default implementation does nothing.

```python
async def initialize(self) -> None:
    pass
```

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

### Monitoring Circuit State with Fallback

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/errors.md

An example function that wraps an API call with circuit breaker logic. It catches OpenedState to log a warning and return cached data, or other exceptions for error logging.

```python
async def call_with_fallback(factory, circuit_name, api_call):
    try:
        breaker = await factory.get_breaker(circuit_name)
        async with breaker:
            return await api_call()
    except OpenedState as e:
        logger.warning(f"Circuit {e.circuit_name} is open. Using cached data.")
        return get_cached_response()
    except Exception as e:
        logger.error(f"Request failed: {e}")
        raise
```

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

### Project Structure Overview

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/README.md

Illustrates the directory structure of the Purgatory project documentation.

```text
README.md                          ← You are here
INDEX.md                           ← Complete navigation
overview.md                        ← Architecture overview
types.md                           ← Type definitions
configuration.md                   ← Configuration reference
errors.md                          ← Exception reference

api-reference/                     ← API documentation
├── AsyncCircuitBreakerFactory.md
├── SyncCircuitBreakerFactory.md
├── AsyncCircuitBreaker.md
├── SyncCircuitBreaker.md
├── Context.md
├── States.md
├── Events.md
├── Commands.md
├── MessageBus.md
├── UnitOfWork.md
└── Repository.md
```

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

### Initialize AsyncCircuitBreakerFactory with In-Memory Persistence

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Instantiate the factory using the default in-memory persistence backend. This is suitable for single-process applications.

```python
factory = AsyncCircuitBreakerFactory()
# Suitable for single-process applications
```

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

### Register Listener for CircuitBreakerCreated Event

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Events.md

Example of how to register a listener function to receive and process CircuitBreakerCreated events. The listener can then access event details like the circuit breaker's name and threshold.

```python
def on_circuit_created(name: str, event_type: str, event: Event) -> None:
    if isinstance(event, CircuitBreakerCreated):
        print(f"Created circuit '{event.name}' with threshold {event.threshold}")

factory.add_listener(on_circuit_created)
```

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

### Context Manager Protocol Methods

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Context.md

Defines the __enter__ and __exit__ methods for synchronous context management. __enter__ is called upon entering the context, and __exit__ is called upon exiting, handling exceptions if they occur.

```python
def __enter__(self) -> "Context"
def __exit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc: Optional[BaseException],
    tb: Optional[TracebackType],
) -> None
```

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

### Get or create an async circuit breaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Retrieves an existing circuit breaker for a given circuit name or creates a new one. Supports overriding default threshold, TTL, and exception exclusion lists.

```python
factory = AsyncCircuitBreakerFactory(default_threshold=3, default_ttl=60)

# Get a breaker and use as context manager
async with await factory.get_breaker("api_call") as breaker:
    result = await external_api_call()

# Override defaults per breaker
async with await factory.get_breaker(
    "slow_api",
    threshold=10,
    ttl=120
) as breaker:
    result = await slow_operation()
```

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

### Configure AsyncCircuitBreakerFactory with Factory Defaults

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/overview.md

Set default failure thresholds and time-to-live (TTL) for circuit breakers. Exclusions and persistence backends can also be configured here.

```python
factory = AsyncCircuitBreakerFactory(
    default_threshold=5,      # Failures before open
    default_ttl=30,           # Seconds before half-open
    exclude=None,             # Exception types to exclude
    uow=None,                 # Persistence backend (in-memory if None)
)
```

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

### Exclude with Custom Predicate

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/AsyncCircuitBreakerFactory.md

Define a custom predicate function to determine if an exception should be excluded. The function should return True to ignore the exception, and False to count it as a failure. This example excludes ValueErrors containing the word 'retry'.

```python
def should_not_fail(exc: BaseException) -> bool:
    # Return False to count as failure, True to ignore
    return isinstance(exc, ValueError) and "retry" in str(exc)

async with await factory.get_breaker(
    "api",
    exclude=[(ValueError, should_not_fail)]
) as breaker:
    await api_call()
```

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

### Override Default Circuit Breaker Configuration

Source: https://github.com/mardiros/purgatory/blob/main/docs/source/user/introduction.md

Use the async context manager to get a circuit breaker, overriding the default threshold and TTL for a specific circuit name. The first circuit to register a name dictates its configuration.

```python
async with await circuitbreaker.get_breaker("www.example.net", threshold=7, ttl=42):
   async with httpx.AsyncClient() as client:
       r = await client.get('https://www.example.net/')

```

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

### Configure AsyncCircuitBreakerFactory with Environment Variables

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

This snippet demonstrates how to configure the `AsyncCircuitBreakerFactory` using environment variables for threshold, TTL, and Redis URL. It dynamically selects between `AsyncRedisUnitOfWork` and `AsyncInMemoryUnitOfWork` based on the `REDIS_URL` environment variable.

```python
import os
from purgatory import AsyncCircuitBreakerFactory

threshold = int(os.getenv("CIRCUIT_BREAKER_THRESHOLD", "5"))
ttl = float(os.getenv("CIRCUIT_BREAKER_TTL", "30"))
redis_url = os.getenv("REDIS_URL")

if redis_url:
    from purgatory import AsyncRedisUnitOfWork
    uow = AsyncRedisUnitOfWork(redis_url)
else:
    from purgatory import AsyncInMemoryUnitOfWork
    uow = AsyncInMemoryUnitOfWork()

factory = AsyncCircuitBreakerFactory(
    default_threshold=threshold,
    default_ttl=ttl,
    uow=uow
)
```

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

### SyncCircuitBreaker.__enter__

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreaker.md

Synchronous context manager entry point. Updates circuit state before executing code by calling context.handle_new_request().

```APIDOC
## __enter__

### Description
Synchronous context manager entry point. Calls `context.handle_new_request()` to update circuit state before executing code.

### Method
`__enter__`

### Parameters
None

### Return
`SyncCircuitBreaker` — self

### Throws
`OpenedState` if circuit is open and TTL has not expired

### Example
```python
factory = SyncCircuitBreakerFactory()
breaker = factory.get_breaker("my_circuit")

with breaker:
    result = api_call()
```
```

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

### Get or Create a SyncCircuitBreaker

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreakerFactory.md

Retrieves an existing circuit breaker for a given circuit name or creates a new one. This method returns the same breaker instance for repeated calls with the same circuit name. Defaults can be overridden per breaker.

```python
factory = SyncCircuitBreakerFactory(default_threshold=3, default_ttl=60)

# Get a breaker and use as context manager
with factory.get_breaker("api_call") as breaker:
    result = external_api_call()

# Override defaults per breaker
with factory.get_breaker(
    "slow_api",
    threshold=10,
    ttl=120
) as breaker:
    result = slow_operation()
```

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

### Exclude with Custom Predicate

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/SyncCircuitBreakerFactory.md

Configure the circuit breaker to exclude exceptions based on a custom predicate function. The function should return True to ignore the exception and False to count it as a failure. This example excludes ValueErrors containing 'retry' in their message.

```python
from purgatory import AuthenticationError

# Exclude with custom predicate
def should_not_fail(exc: BaseException) -> bool:
    # Return False to count as failure, True to ignore
    return isinstance(exc, ValueError) and "retry" in str(exc)

with factory.get_breaker(
    "api",
    exclude=[(ValueError, should_not_fail)]
) as breaker:
    api_call()
```

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

### Context Manager Protocol

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Implements the synchronous context manager protocol (__enter__ and __exit__). On exception, it calls the rollback method. __enter__ returns the SyncAbstractUnitOfWork instance.

```python
def __enter__(self) -> SyncAbstractUnitOfWork
def __exit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc: Optional[BaseException],
    tb: Optional[TracebackType],
) -> None
```

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

### AsyncRedisUnitOfWork Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/UnitOfWork.md

Initializes a Redis-backed UnitOfWork for distributed systems. Requires a Redis connection URL.

```APIDOC
## AsyncRedisUnitOfWork Constructor

### Description
Initializes a Redis-backed UnitOfWork for distributed systems. Requires a Redis connection URL.

### Constructor
```python
class AsyncRedisUnitOfWork(AsyncAbstractUnitOfWork):
    def __init__(self, url: str) -> None
```

### Parameters
#### Path Parameters
- **url** (str) - Required - Redis connection URL (e.g., "redis://localhost:6379")

### Attributes
- `contexts` (AsyncRedisRepository) — Redis-backed storage

### Throws
- `ConfigurationError` if redis package is not installed
```

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

### Context Constructor

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/api-reference/Context.md

Initializes a new Context object for a circuit breaker. Use this to set up the circuit breaker's name, failure threshold, recovery time, and initial state.

```python
Context(
    name: CircuitName,
    threshold: Threshold,
    ttl: TTL,
    state: StateName = "closed",
    failure_count: int = 0,
    opened_at: Optional[float] = None,
    exclude: Optional[ExcludeType] = None,
)
```

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

### Async In-Memory Storage

Source: https://github.com/mardiros/purgatory/blob/main/_autodocs/configuration.md

Default configuration for Purgatory's async circuit breaker factory, utilizing in-memory storage.

```python
from purgatory import AsyncCircuitBreakerFactory

factory = AsyncCircuitBreakerFactory()
# Automatically uses AsyncInMemoryUnitOfWork()
```