Fluxgate - Python-Circuit-Breaker mit Unterstützung für Sliding Windows und kombinierbare Regeln

(byexist.github.io)

3 Punkte von byexist 2026-01-05 | Noch keine Kommentare. | Auf WhatsApp teilen

Bei der Nutzung bestehender Python-Circuit-Breaker-Bibliotheken wie circuitbreaker oder pybreaker gab es einige Punkte, die ich vermisst habe.

Es wird nur ein Zähler für aufeinanderfolgende Fehler verwendet: Ein einzelner Erfolg setzt ihn zurück, wodurch instabile Services schwer zu erkennen sind
Bei der Wiederherstellung ist nur ein einzelner Testaufruf erlaubt: Für Dinge wie Cache-Warm-up ist das unzureichend

Die Unterstützung von Django-Permission-Operatoren und das Window-Management von resilience4j haben mich inspiriert.

Installation

pip install fluxgate

Grundlegende Verwendung

from fluxgate import CircuitBreaker  
  
cb = CircuitBreaker(name="external_api")  
  
@cb  
def call_api():  
    return requests.get("https://api.example.com/data";)

Standardwert: Auslösung bei 50 % Fehlern in den letzten 100 Aufrufen, Wiederherstellungsversuch nach 60 Sekunden

Unterstützung für asynchrone Nutzung

from fluxgate import AsyncCircuitBreaker  
  
cb = AsyncCircuitBreaker(name="external_api")  
  
@cb  
async def call_api():  
    async with httpx.AsyncClient() as client:  
        return await client.get("https://api.example.com/data";)

Zentrale Funktionen

Sliding Window + Auslösung auf Basis der Fehlerrate

from fluxgate import CircuitBreaker  
from fluxgate.trippers import MinRequests, FailureRate, FailureStreak  
  
cb = CircuitBreaker(  
    name="api",  
    tripper=FailureStreak(5) | (MinRequests(100) & FailureRate(0.5))  
)

→ Auslösung bei 5 aufeinanderfolgenden Fehlern oder wenn 50 % der letzten 100 Aufrufe fehlschlagen

Schrittweise Wiederherstellung (RampUp)

from fluxgate.permits import RampUp  
  
cb = CircuitBreaker(  
    name="api",  
    permit=RampUp(initial=0.1, final=1.0, duration=60.0)  
)

→ Bei der Wiederherstellung steigt der Traffic über 60 Sekunden schrittweise von 10 % auf 100 %

Funktionsvergleich mit bestehenden Bibliotheken

Funktion	fluxgate	circuitbreaker	pybreaker	aiobreaker
Sliding Window	O	X	X	X
Auslösung nach Fehlerrate	O	X	X	X
Kombinierbare Regeln	O	X	X	X
Schrittweise Wiederherstellung	O	X	X	X