Circuit Breaker

Important

The synchronous version is generated from the async version

Async

class purgatory.service._async.circuitbreaker.AsyncCircuitBreaker(context: Context, uow: AsyncAbstractUnitOfWork, messagebus: AsyncMessageRegistry)

class purgatory.service._async.circuitbreaker.PublicEvent(messagebus: AsyncMessageRegistry, hook: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None])

remove_listeners(messagebus: AsyncMessageRegistry) → None

async cb_created(event: CircuitBreakerCreated, uow: AsyncAbstractUnitOfWork) → None

async cb_state_changed(event: CircuitBreakerCreated, uow: AsyncAbstractUnitOfWork) → None

async cb_failed(event: CircuitBreakerCreated, uow: AsyncAbstractUnitOfWork) → None

async cb_recovered(event: CircuitBreakerCreated, uow: AsyncAbstractUnitOfWork) → None

class purgatory.service._async.circuitbreaker.AsyncCircuitBreakerFactory(default_threshold: int = 5, default_ttl: float = 30, exclude: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]] | None = None, uow: AsyncAbstractUnitOfWork | None = None)

async initialize() → None

add_listener(listener: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None]) → None

remove_listener(listener: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None]) → None

async get_breaker(circuit: str, threshold: int | None = None, ttl: float | None = None, exclude: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]] | None = None) → AsyncCircuitBreaker

Sync

class purgatory.service._sync.circuitbreaker.SyncCircuitBreaker(context: Context, uow: SyncAbstractUnitOfWork, messagebus: SyncMessageRegistry)

class purgatory.service._sync.circuitbreaker.PublicEvent(messagebus: SyncMessageRegistry, hook: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None])

remove_listeners(messagebus: SyncMessageRegistry) → None

cb_created(event: CircuitBreakerCreated, uow: SyncAbstractUnitOfWork) → None

cb_state_changed(event: CircuitBreakerCreated, uow: SyncAbstractUnitOfWork) → None

cb_failed(event: CircuitBreakerCreated, uow: SyncAbstractUnitOfWork) → None

cb_recovered(event: CircuitBreakerCreated, uow: SyncAbstractUnitOfWork) → None

class purgatory.service._sync.circuitbreaker.SyncCircuitBreakerFactory(default_threshold: int = 5, default_ttl: float = 30, exclude: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]] | None = None, uow: SyncAbstractUnitOfWork | None = None)

initialize() → None

add_listener(listener: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None]) → None

remove_listener(listener: Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], Event], None]) → None

get_breaker(circuit: str, threshold: int | None = None, ttl: float | None = None, exclude: list[type[BaseException] | tuple[type[BaseException], Callable[[...], bool]]] | None = None) → SyncCircuitBreaker