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: typing_extensions.Literal[opened, closed, half - opened] = 'closed', failure_count: int = 0, opened_at: Optional[float] = None, exclude: Optional[List[Union[Type[BaseException], Tuple[Type[BaseException], Callable[[...], bool]]]]] = None)

name: str

ttl: float

threshold: int

messages: List[purgatory.domain.messages.base.Event]

exclude_list: List[Union[Type[BaseException], Tuple[Type[BaseException], Callable[[...], bool]]]]

property state: typing_extensions.Literal[opened, closed, half-opened]

property opened_at: Optional[float]

property failure_count: Optional[int]

set_state(state: purgatory.domain.model.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: Optional[int] = None

opened_at: Optional[float] = None

name: str = ''

abstract handle_new_request(context: purgatory.domain.model.Context) → None

Handle new code block

abstract handle_exception(context: purgatory.domain.model.Context, exc: BaseException) → None

Handle exception during the code block

abstract handle_end_request(context: purgatory.domain.model.Context) → None

Handle proper execution after the code block

class purgatory.domain.model.ClosedState

In closed state, track for failure.

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

handle_new_request(context: purgatory.domain.model.Context) → None

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

handle_exception(context: purgatory.domain.model.Context, exc: BaseException) → None

Handle exception during the code block

handle_end_request(context: purgatory.domain.model.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: typing_extensions.Literal[opened, closed, half-opened] = 'opened'

handle_new_request(context: purgatory.domain.model.Context) → None

Handle new code block

handle_exception(context: purgatory.domain.model.Context, exc: BaseException) → None

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

This function is never called.

handle_end_request(context: purgatory.domain.model.Context) → None

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

This function is never called.

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

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

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

handle_new_request(context: purgatory.domain.model.Context) → None

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

handle_exception(context: purgatory.domain.model.Context, exc: BaseException) → None

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

handle_end_request(context: purgatory.domain.model.Context) → None

Otherwise, the circuit is closed, back to normal.