Fluxgate - un circuit breaker Python avec prise en charge de règles combinables et de fenêtres glissantes

En utilisant les bibliothèques Python de circuit breaker existantes (circuitbreaker, pybreaker, etc.), j’ai trouvé certains points frustrants.

• Utilisation uniquement d’un compteur d’échecs consécutifs : une seule réussite réinitialise le compteur, ce qui rend difficile la détection de services instables
• Lors de la reprise, un seul appel de test est autorisé : insuffisant pour des cas comme le réchauffement du cache

Je me suis inspiré de la prise en charge des opérateurs de permissions de Django et de la gestion des fenêtres de resilience4j.

Installation

pip install fluxgate  

Utilisation de base

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

Valeurs par défaut : déclenchement si 50 % des 100 appels les plus récents échouent, puis tentative de reprise après 60 secondes

Prise en charge de l’asynchrone

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";)  

Fonctionnalités clés

Fenêtre glissante + déclenchement basé sur le taux d’échec

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

→ Déclenchement après 5 échecs consécutifs, ou si 50 % des 100 appels les plus récents échouent

Reprise progressive (RampUp)

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

→ Lors de la reprise, le trafic augmente progressivement de 10 % à 100 % sur 60 secondes

Comparaison des fonctionnalités avec les bibliothèques existantes

Fonctionnalité	fluxgate	circuitbreaker	pybreaker	aiobreaker
Fenêtre glissante	O	X	X	X
Déclenchement par taux d’échec	O	X	X	X
Règles combinables	O	X	X	X
Reprise progressive	O	X	X	X

Liens

• GitHub : https://github.com/byExist/fluxgate
• Documentation : https://byExist.github.io/fluxgate/

Tous vos retours sont les bienvenus !