API: Interfaces

This page documents the core interfaces that define the behavior of Fluxgate's components. You can implement these interfaces to create your own custom components (windows, trackers, trippers, retries, permits, and listeners).

---

fluxgate.windows.Window

Bases: ABC

Base class for sliding windows over recent call records.

Subclasses must override record, get_metric, and reset.

record abstractmethod

record(record: Record) -> None

Source code in fluxgate/windows.py

@abstractmethod
def record(self, record: Record) -> None: ...

get_metric abstractmethod

get_metric() -> Metric

Source code in fluxgate/windows.py

@abstractmethod
def get_metric(self) -> Metric: ...

reset abstractmethod

reset() -> None

Source code in fluxgate/windows.py

@abstractmethod
def reset(self) -> None: ...

fluxgate.trackers.Tracker

Bases: ABC

Base class for exception trackers.

Subclasses override __call__. The &/|/~ operators build composite trackers automatically.

__call__ abstractmethod

__call__(exception: Exception) -> bool

Source code in fluxgate/trackers.py

@abstractmethod
def __call__(self, exception: Exception) -> bool: ...

__and__

__and__(other: Tracker) -> Tracker

Source code in fluxgate/trackers.py

def __and__(self, other: Tracker) -> Tracker:
    return _And(self, other)

__or__

__or__(other: Tracker) -> Tracker

Source code in fluxgate/trackers.py

def __or__(self, other: Tracker) -> Tracker:
    return _Or(self, other)

__invert__

__invert__() -> Tracker

Source code in fluxgate/trackers.py

def __invert__(self) -> Tracker:
    return _Not(self)

fluxgate.trippers.Tripper

Bases: ABC

Base class for tripper conditions.

Subclasses must override __call__. Composite trippers (_And/_Or) override __iter__ to expose their children; a leaf yields itself by default. The & and | operators build composite trippers.

__call__ abstractmethod

__call__(ctx: CallContext) -> bool

Source code in fluxgate/trippers.py

@abstractmethod
def __call__(self, ctx: CallContext) -> bool: ...

__iter__

__iter__() -> Iterator[Tripper]

Source code in fluxgate/trippers.py

def __iter__(self) -> Iterator[Tripper]:
    yield self

__and__

__and__(other: Tripper) -> Tripper

Source code in fluxgate/trippers.py

def __and__(self, other: Tripper) -> Tripper:
    return _And(self, other)

__or__

__or__(other: Tripper) -> Tripper

Source code in fluxgate/trippers.py

def __or__(self, other: Tripper) -> Tripper:
    return _Or(self, other)

fluxgate.retries.Retry

Bases: ABC

Base class for retry strategies.

Subclasses override __call__ to decide whether the breaker should transition from OPEN to HALF_OPEN at the current moment.

__call__ abstractmethod

__call__(changed_at: float, reopens: int) -> bool

Source code in fluxgate/retries.py

@abstractmethod
def __call__(self, changed_at: float, reopens: int) -> bool: ...

fluxgate.permits.Permit

Bases: ABC

Base class for permit strategies.

Subclasses override __call__ to decide whether a probe call should be admitted while the breaker is in HALF_OPEN.

__call__ abstractmethod

__call__(changed_at: float) -> bool

Source code in fluxgate/permits.py

@abstractmethod
def __call__(self, changed_at: float) -> bool: ...

fluxgate.listeners.Listener

Bases: Protocol

Protocol for synchronous circuit breaker listeners.

Any callable with the matching signature satisfies this protocol — a plain function or a class implementing __call__.

__call__

__call__(signal: Signal) -> None

Source code in fluxgate/listeners/base.py

def __call__(self, signal: Signal, /) -> None: ...

fluxgate.listeners.AsyncListener

Bases: Protocol

Protocol for asynchronous circuit breaker listeners.

__call__ async

__call__(signal: Signal) -> None

Source code in fluxgate/listeners/base.py

async def __call__(self, signal: Signal, /) -> None: ...