API: Permits

This page documents the available recovery permit strategies. Permit strategies control how many "probe" calls are allowed to pass through during the HALF_OPEN state to test for service recovery.

For a high-level guide on choosing a permit strategy, see the Permits Component Guide.

---

fluxgate.permits.All

Bases: IPermit

Permit strategy that always allows calls.

Useful for testing or when you want all calls to pass through in HALF_OPEN state.

Examples:

>>> permit = All()

__call__

__call__(_changed_at: float) -> bool

Source code in fluxgate/permits.py

def __call__(self, _changed_at: float) -> bool:
    return True

fluxgate.permits.Random

Random(ratio: float)

Bases: IPermit

Permit strategy with random sampling in HALF_OPEN state.

Allows calls randomly based on a fixed probability ratio. Simple and effective for limiting concurrent calls during recovery.

Examples:

>>> # Allow 50% of calls in HALF_OPEN state
>>> permit = Random(ratio=0.5)

Parameters:

Name	Type	Description	Default
ratio	float	Probability of allowing a call (0.0 to 1.0)	required

Source code in fluxgate/permits.py

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

__call__

__call__(_changed_at: float) -> bool

Source code in fluxgate/permits.py

def __call__(self, _changed_at: float) -> bool:
    return random() < self._ratio

fluxgate.permits.RampUp

RampUp(initial: float, final: float, duration: float)

Bases: IPermit

Permit strategy that gradually increases allowed traffic over time.

Starts with a low permit ratio and gradually increases to the final ratio over the specified duration. Useful for smooth recovery without sudden load spikes.

Examples:

>>> # Start at 10%, ramp up to 80% over 60 seconds
>>> permit = RampUp(initial=0.1, final=0.8, duration=60.0)

Parameters:

Name	Type	Description	Default
initial	float	Initial permit ratio (0.0 to 1.0)	required
final	float	Final permit ratio (0.0 to 1.0, must be >= initial)	required
duration	float	Ramp-up duration in seconds	required

Source code in fluxgate/permits.py

def __init__(self, initial: float, final: float, duration: float) -> None:
    if not (0.0 <= initial <= final <= 1.0):
        raise ValueError("Initial and final must be between 0.0 and 1.0")
    if duration <= 0:
        raise ValueError("Duration must be greater than zero")
    self._initial = initial
    self._final = final
    self._duration = duration

__call__

__call__(changed_at: float) -> bool

Source code in fluxgate/permits.py

def __call__(self, changed_at: float) -> bool:
    elapsed = time() - changed_at
    if elapsed >= self._duration:
        return random() < self._final
    ratio = self._initial + (self._final - self._initial) * (
        elapsed / self._duration
    )
    return random() < ratio