Config

Configuration for the Circuit Breaker utility.

CLASS	DESCRIPTION
CircuitBreakerConfig	Tunables for a circuit breaker.

CircuitBreakerConfig

CircuitBreakerConfig(failure_threshold: int = 5, recovery_timeout: int = 30, success_threshold: int = 3, handled_exceptions: tuple[type[Exception], ...] | None = None, ignored_exceptions: tuple[type[Exception], ...] | None = None, local_cache_max_age: int = 5)

Tunables for a circuit breaker.

All values have sensible defaults, so CircuitBreakerConfig() is a valid production configuration. Pass an instance to @circuit_breaker(config=...) to override them.

PARAMETER	DESCRIPTION
failure_threshold	Number of consecutive failures that trips a closed circuit to open. Defaults to 5. TYPE: int DEFAULT: 5
recovery_timeout	Seconds the circuit stays open before allowing a half-open probe. Defaults to 30. TYPE: int DEFAULT: 30
success_threshold	Number of consecutive probe successes required to close a half-open circuit. Defaults to 3. TYPE: int DEFAULT: 3
handled_exceptions	propagates without affecting the circuit. Mutually exclusive with ignored_exceptions. Defaults to None (treated as (Exception,)). TYPE: tuple[type[Exception], ...] | None DEFAULT: None
ignored_exceptions	exclusive with handled_exceptions. Defaults to None. TYPE: tuple[type[Exception], ...] | None DEFAULT: None
local_cache_max_age	Seconds a circuit's state is cached in the execution environment before a read-through to the store. Matches the Parameters utility default. Defaults to 5. TYPE: int DEFAULT: 5

RAISES	DESCRIPTION
CircuitBreakerConfigError	If both handled_exceptions and ignored_exceptions are provided, or a numeric tunable is not a positive integer.

Example

Only count timeouts and connection errors as failures

config = CircuitBreakerConfig(
    failure_threshold=5,
    recovery_timeout=30,
    handled_exceptions=(TimeoutError, ConnectionError),
)

METHOD	DESCRIPTION
counts_as_failure	Decide whether an exception raised by the protected call counts as a circuit failure.

Source code in aws_lambda_powertools/utilities/circuit_breaker_alpha/config.py

def __init__(
    self,
    failure_threshold: int = 5,
    recovery_timeout: int = 30,
    success_threshold: int = 3,
    handled_exceptions: tuple[type[Exception], ...] | None = None,
    ignored_exceptions: tuple[type[Exception], ...] | None = None,
    local_cache_max_age: int = 5,
):
    self._validate(
        failure_threshold=failure_threshold,
        recovery_timeout=recovery_timeout,
        success_threshold=success_threshold,
        handled_exceptions=handled_exceptions,
        ignored_exceptions=ignored_exceptions,
        local_cache_max_age=local_cache_max_age,
    )

    self.failure_threshold = failure_threshold
    self.recovery_timeout = recovery_timeout
    self.success_threshold = success_threshold
    self.handled_exceptions = handled_exceptions
    self.ignored_exceptions = ignored_exceptions
    self.local_cache_max_age = local_cache_max_age

counts_as_failure

counts_as_failure(exception: Exception) -> bool

Decide whether an exception raised by the protected call counts as a circuit failure.

PARAMETER	DESCRIPTION
exception	The exception raised by the protected function. TYPE: Exception

RETURNS	DESCRIPTION
bool	True if the exception should increment the failure counter, False if it should propagate without affecting the circuit.

Source code in aws_lambda_powertools/utilities/circuit_breaker_alpha/config.py

def counts_as_failure(self, exception: Exception) -> bool:
    """
    Decide whether an exception raised by the protected call counts as a circuit failure.

    Parameters
    ----------
    exception : Exception
        The exception raised by the protected function.

    Returns
    -------
    bool
        ``True`` if the exception should increment the failure counter, ``False`` if
        it should propagate without affecting the circuit.
    """
    if self.handled_exceptions is not None:
        return isinstance(exception, self.handled_exceptions)
    if self.ignored_exceptions is not None:
        return not isinstance(exception, self.ignored_exceptions)
    # Default: any exception counts as a failure.
    return True