API: Trippers

This page documents the available tripping strategies. Trippers are responsible for evaluating metrics from a window to decide if the circuit should change state.

For a high-level guide on choosing and combining trippers, see the Trippers Component Guide.

---

fluxgate.trippers.Closed

Bases: TripperBase

Tripper that returns true only when circuit is in CLOSED state.

Used to compose conditions that should only apply when circuit is closed.

Examples:

>>> # Only check failure rate when circuit is closed
>>> tripper = Closed() & FailureRate(0.5)

__call__

__call__(
    _metric: Metric,
    state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, _metric: Metric, state: StateEnum, _consecutive_failures: int
) -> bool:
    return state == StateEnum.CLOSED

fluxgate.trippers.HalfOpened

Bases: TripperBase

Tripper that returns true only when circuit is in HALF_OPEN state.

Used to compose conditions that should only apply when circuit is half-open.

Examples:

>>> # Only check failure rate when circuit is half-open
>>> tripper = HalfOpened() & FailureRate(0.3)

__call__

__call__(
    _metric: Metric,
    state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, _metric: Metric, state: StateEnum, _consecutive_failures: int
) -> bool:
    return state == StateEnum.HALF_OPEN

fluxgate.trippers.MinRequests

MinRequests(count: int)

Bases: TripperBase

Tripper that requires minimum number of calls before evaluating.

Prevents premature circuit opening when sample size is too small.

Examples:

>>> # Only trip after at least 10 calls
>>> tripper = MinRequests(10) & FailureRate(0.5)

Parameters:

Name	Type	Description	Default
count	int	Minimum number of calls required	required

Source code in fluxgate/trippers.py

def __init__(self, count: int) -> None:
    if count <= 0:
        raise ValueError("Count must be greater than zero")
    self._count = count

__call__

__call__(
    metric: Metric,
    _state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, metric: Metric, _state: StateEnum, _consecutive_failures: int
) -> bool:
    return metric.total_count >= self._count

fluxgate.trippers.FailureRate

FailureRate(ratio: float)

Bases: TripperBase

Tripper based on failure rate threshold.

Returns true when the ratio of failed calls exceeds the threshold.

Examples:

>>> # Trip when 50% or more calls fail
>>> tripper = FailureRate(0.5)

Parameters:

Name	Type	Description	Default
ratio	float	Failure rate threshold (0.0 to 1.0)	required

Source code in fluxgate/trippers.py

def __init__(self, ratio: float) -> None:
    if ratio <= 0 or ratio > 1:
        raise ValueError("Ratio must be between 0 and 1")
    self._ratio = ratio

__call__

__call__(
    metric: Metric,
    _state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, metric: Metric, _state: StateEnum, _consecutive_failures: int
) -> bool:
    if metric.total_count == 0:
        return False
    failure_rate = metric.failure_count / metric.total_count
    return failure_rate >= self._ratio

fluxgate.trippers.AvgLatency

AvgLatency(threshold: float)

Bases: TripperBase

Tripper based on average latency threshold.

Returns true when average call duration reaches or exceeds the threshold.

Examples:

>>> # Trip when average latency reaches 2 seconds
>>> tripper = AvgLatency(2.0)

Parameters:

Name	Type	Description	Default
threshold	float	Latency threshold in seconds	required

Source code in fluxgate/trippers.py

def __init__(self, threshold: float) -> None:
    if threshold <= 0:
        raise ValueError("Threshold must be greater than 0")
    self._threshold = threshold

__call__

__call__(
    metric: Metric,
    _state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, metric: Metric, _state: StateEnum, _consecutive_failures: int
) -> bool:
    if metric.total_count == 0:
        return False
    avg_duration = metric.total_duration / metric.total_count
    return avg_duration >= self._threshold

fluxgate.trippers.SlowRate

SlowRate(ratio: float)

Bases: TripperBase

Tripper based on slow call rate threshold.

Returns true when the ratio of slow calls exceeds the threshold. Slow calls are determined by the circuit breaker's slow_threshold.

Examples:

>>> # Trip when 30% or more calls are slow
>>> tripper = SlowRate(0.3)

Parameters:

Name	Type	Description	Default
ratio	float	Slow call rate threshold (0.0 to 1.0)	required

Source code in fluxgate/trippers.py

def __init__(self, ratio: float) -> None:
    if ratio <= 0 or ratio > 1:
        raise ValueError("Ratio must be between 0 and 1")
    self._ratio = ratio

__call__

__call__(
    metric: Metric,
    _state: StateEnum,
    _consecutive_failures: int,
) -> bool

Source code in fluxgate/trippers.py

def __call__(
    self, metric: Metric, _state: StateEnum, _consecutive_failures: int
) -> bool:
    if metric.total_count == 0:
        return False
    slow_rate = metric.slow_count / metric.total_count
    return slow_rate >= self._ratio