Models

Circuit breaker state model.

The state model is implemented using the State pattern from the Gang Of Four.

class purgatory.domain.model.Context(name: str, threshold: int, ttl: float, state: Literal['opened', 'closed', 'half-opened'] = 'closed', failure_count: int = 0, opened_at: float | None = None, exclude: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]] | None = None)

name: str

ttl: float

threshold: int

messages: list[Event]

exclude_list: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]]

property state: Literal['opened', 'closed', 'half-opened']

property opened_at: float | None

property failure_count: int | None

set_state(state: State) → None

mark_failure(failure_count: int) → None

recover_failure() → None

handle_new_request() → None

handle_exception(exc: BaseException) → None

handle_end_request() → None

class purgatory.domain.model.State

failure_count: int | None = None

opened_at: float | None = None

name: str = ''

abstract handle_new_request(context: Context) → None

Handle new code block

abstract handle_exception(context: Context, exc: BaseException) → None

Handle exception during the code block

abstract handle_end_request(context: Context) → None

Handle proper execution after the code block

class purgatory.domain.model.ClosedState

In closed state, track for failure.

name: Literal['opened', 'closed', 'half-opened'] = 'closed'

handle_new_request(context: Context) → None

When the circuit is closed, the new request has no incidence

handle_exception(context: Context, exc: BaseException) → None

Handle exception during the code block

handle_end_request(context: Context) → None

Reset in case the request is ok

exception purgatory.domain.model.OpenedState(circuit_name: str)

In open state, reopen after a TTL.

name: Literal['opened', 'closed', 'half-opened'] = 'opened'

handle_new_request(context: Context) → None

Handle new code block

handle_exception(context: Context, exc: BaseException) → None

When the circuit is opened, the OpenState is raised before entering.

This function is never called.

handle_end_request(context: Context) → None

When the circuit is opened, the OpenState is raised before entering.

This function is never called.

class purgatory.domain.model.HalfOpenedState(name: Literal['opened', 'closed', 'half-opened'] = 'half-opened')

In half open state, decide to reopen or to close.

name: Literal['opened', 'closed', 'half-opened'] = 'half-opened'

handle_new_request(context: Context) → None

In half open state, we reset the failure counter to restart 0.

handle_exception(context: Context, exc: BaseException) → None

If an exception happens, then the circuit is reopen directly.

handle_end_request(context: Context) → None

Otherwise, the circuit is closed, back to normal.