API Reference

CircuitProtectorPolicy

class resilient_circuit.CircuitProtectorPolicy[source]

Bases: ProtectionPolicy

DEFAULT_THRESHOLD = Fraction(1, 1)

__init__(*, resource_key=None, storage=None, namespace=None, cooldown=datetime.timedelta(0), failure_limit=Fraction(1, 1), success_limit=Fraction(1, 1), should_handle=<function CircuitProtectorPolicy.<lambda>>, on_status_change=None)[source]

Parameters:

resource_key (str | None)
storage (CircuitBreakerStorage | None)
namespace (str | None)
cooldown (timedelta)
failure_limit (Fraction)
success_limit (Fraction)
should_handle (Callable[[Exception], bool])
on_status_change (Callable[[CircuitProtectorPolicy, CircuitStatus, CircuitStatus], None] | None)

Return type:

None

property execution_log: BinaryCircularBuffer

property status: CircuitStatus

on_status_change(current, new)[source]

This method is called whenever protector changes its status.

Parameters:

current (CircuitStatus)
new (CircuitStatus)

Return type:

None

CircuitStatus

RetryWithBackoffPolicy

class resilient_circuit.RetryWithBackoffPolicy[source]

Bases: ProtectionPolicy

__init__(*, backoff=None, max_retries=3, should_handle=<function RetryWithBackoffPolicy.<lambda>>)[source]

Parameters:

backoff (ExponentialDelay | None)
max_retries (int)
should_handle (Callable[[Exception], bool])

SafetyNet

class resilient_circuit.SafetyNet[source]

Bases: object

Decorates function with given policies.

SafetyNet will handle execution results in reverse, with last policy applied first.

Example

>>> from resilient_circuit import SafetyNet, RetryWithBackoffPolicy, CircuitProtectorPolicy
>>>
>>> @SafetyNet(policies=(RetryWithBackoffPolicy(), CircuitProtectorPolicy()))
>>> def some_method() -> bool:
>>>     return True

__init__(*, policies)[source]

Parameters:: policies (Sequence[ProtectionPolicy])
Return type:: None

__call__(func)[source]

Decorate func with all policies in reversed order.

Parameters:: func (Callable[[~P], R])
Return type:: Callable[[~P], R]

ExponentialDelay

class resilient_circuit.ExponentialDelay[source]

Bases: object

ExponentialDelay(min_delay: datetime.timedelta, max_delay: datetime.timedelta, factor: int = 2, jitter: Optional[float] = None)

min_delay: timedelta

max_delay: timedelta

factor: int = 2

jitter: float | None = None

for_attempt(attempt)[source]

Compute delay in seconds for a given attempt.

Parameters:: attempt (int)
Return type:: float

__init__(min_delay, max_delay, factor=2, jitter=None)

Parameters:

min_delay (timedelta)
max_delay (timedelta)
factor (int)
jitter (float | None)

Return type:

None

FixedDelay

class resilient_circuit.FixedDelay[source]

Bases: ExponentialDelay

Special case of ExponentialDelay when delay between calls is constant.

__init__(delay)[source]

Parameters:: delay (timedelta)
Return type:: None

Storage Classes

class resilient_circuit.storage.CircuitBreakerStorage[source]

Bases: ABC

Abstract base class for circuit breaker storage backends.

abstractmethod get_state(resource_key)[source]

Get the state for a given resource key.

Returns:: state, failure_count, open_until, execution_log (optional) or None if no state found
Return type:: Dictionary with keys
Parameters:: resource_key (str)

abstractmethod set_state(resource_key, state, failure_count, open_until, execution_log=None)[source]

Set the state for a given resource key.

Parameters:

execution_log (list | None) – Optional list of boolean success/failure results
resource_key (str)
state (str)
failure_count (int)
open_until (float)

Return type:

None

class resilient_circuit.storage.InMemoryStorage[source]

Bases: CircuitBreakerStorage

In-memory storage implementation for circuit breaker state.

__init__()[source]

Return type:: None

get_state(resource_key)[source]

Get the state for a given resource key.

Returns:: state, failure_count, open_until, execution_log (optional) or None if no state found
Return type:: Dictionary with keys
Parameters:: resource_key (str)

set_state(resource_key, state, failure_count, open_until, execution_log=None)[source]

Set the state for a given resource key.

Parameters:

execution_log (list | None) – Optional list of boolean success/failure results
resource_key (str)
state (str)
failure_count (int)
open_until (float)

Return type:

None

class resilient_circuit.storage.PostgresStorage[source]

Bases: CircuitBreakerStorage

PostgreSQL storage implementation for circuit breaker state.

__init__(connection_string, namespace='default')[source]

Parameters:

connection_string (str)
namespace (str)

get_state(resource_key)[source]

Get the state for a given resource key within this namespace.

NOTE: This query uses FOR UPDATE to lock the row to ensure this read-call-write cycle is atomic. Namespace isolation ensures parallel tests don’t conflict.

Parameters:: resource_key (str)
Return type:: Dict[str, Any] | None

set_state(resource_key, state, failure_count, open_until, execution_log=None)[source]

Set the state for a given resource key within this namespace.

Parameters:

resource_key (str) – Unique circuit breaker identifier
state (str) – Circuit state (CLOSED, OPEN, HALF_OPEN)
failure_count (int) – Number of consecutive failures
open_until (float) – Timestamp when circuit can transition from OPEN
execution_log (list | None) – Optional list of boolean success/failure results for the circular buffer

Return type:

None