API: Tripper

이 페이지는 사용 가능한 Tripper를 문서화합니다. Tripper는 window의 Metric을 평가하여 회로가 상태를 변경해야 하는지 결정하는 역할을 합니다.

Tripper 선택 및 조합에 대한 개요는 Tripper 컴포넌트 가이드를 참조하십시오.

---

fluxgate.trippers.CallContext dataclass

CallContext(
    *,
    metric: Metric,
    state: State,
    consecutive_failures: int,
)

Snapshot passed to trippers on each evaluation.

Carries the window-derived :class:Metric together with the breaker's immediate state and consecutive failure count.

metric instance-attribute

metric: Metric

state instance-attribute

state: State

consecutive_failures instance-attribute

consecutive_failures: int

fluxgate.trippers.Closed

Bases: Tripper

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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    return ctx.state == "closed"

fluxgate.trippers.HalfOpened

Bases: Tripper

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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    return ctx.state == "half_open"

fluxgate.trippers.MinRequests

MinRequests(count: int)

Bases: Tripper

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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    return ctx.metric.total_count >= self._count

fluxgate.trippers.FailureRate

FailureRate(ratio: float)

Bases: Tripper

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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    if ctx.metric.total_count == 0:
        return False
    failure_rate = ctx.metric.failure_count / ctx.metric.total_count
    return failure_rate >= self._ratio

fluxgate.trippers.AvgLatency

AvgLatency(threshold: float)

Bases: Tripper

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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    if ctx.metric.total_count == 0:
        return False
    avg_duration = ctx.metric.total_duration / ctx.metric.total_count
    return avg_duration >= self._threshold

fluxgate.trippers.SlowRate

SlowRate(ratio: float, *, threshold: float)

Bases: Tripper

Tripper based on slow call rate against a duration threshold.

A call is "slow" when its duration is at least threshold seconds. The tripper fires when the ratio of slow calls in the current window reaches ratio. Several SlowRate instances with different thresholds can coexist in the same tripper tree — the circuit breaker walks the tree and registers each threshold with the window.

Examples:

>>> # Trip when 30% of calls take 1s or longer.
>>> tripper = SlowRate(0.3, threshold=1.0)
>>>
>>> # Combine layered thresholds for tiered alerts.
>>> tripper = SlowRate(0.3, threshold=1.0) | SlowRate(0.05, threshold=10.0)

Parameters:

Name	Type	Description	Default
ratio	float	Slow call rate threshold (0.0 to 1.0)	required
threshold	float	Duration in seconds at or above which a call counts as slow	required

Source code in fluxgate/trippers.py

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

threshold instance-attribute

threshold = threshold

__call__

__call__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    if ctx.metric.total_count == 0:
        return False
    slow_count = ctx.metric.slow_counts.get(self.threshold, 0)
    return slow_count / ctx.metric.total_count >= self._ratio

fluxgate.trippers.FailureStreak

FailureStreak(count: int)

Bases: Tripper

Tripper based on consecutive failure count.

Returns true when the number of consecutive failures reaches the threshold. Useful for fast failure detection during cold start or when external service is completely down.

Examples:

>>> # Trip after 5 consecutive failures
>>> tripper = FailureStreak(5)
>>>
>>> # Combine with other trippers for comprehensive protection
>>> tripper = FailureStreak(5) | (MinRequests(20) & FailureRate(0.5))

Parameters:

Name	Type	Description	Default
count	int	Number of consecutive failures required to trip	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__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

def __call__(self, ctx: CallContext) -> bool:
    return ctx.consecutive_failures >= self._count