Fluxgate - Circuit Breaker para Python con soporte para reglas combinables y ventana deslizante

Había algunos puntos que me dejaban insatisfecho al usar bibliotecas existentes de Circuit Breaker para Python (circuitbreaker, pybreaker, etc.).

• Solo usan conteo de fallos consecutivos: si una llamada tiene éxito, se reinicia, lo que dificulta detectar servicios inestables
• Al recuperarse, solo permiten una llamada de prueba: insuficiente para casos como calentamiento de caché

Me inspiré en el soporte de operadores de Django Permission y en la gestión de ventanas de resilience4j.

Instalación

pip install fluxgate  

Uso básico

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

Valor predeterminado: se activa si falla el 50% de las últimas 100 llamadas e intenta recuperarse después de 60 segundos

Soporte asíncrono

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

Funciones clave

Ventana deslizante + activación basada en tasa de fallos

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

→ Se activa tras 5 fallos consecutivos o si falla el 50% de las últimas 100 llamadas

Recuperación gradual (RampUp)

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

→ Durante la recuperación, el tráfico aumenta gradualmente del 10% al 100% en 60 segundos

Comparación de funciones con bibliotecas existentes

Función	fluxgate	circuitbreaker	pybreaker	aiobreaker
Ventana deslizante	O	X	X	X
Activación por tasa de fallos	O	X	X	X
Reglas combinables	O	X	X	X
Recuperación gradual	O	X	X	X

Enlaces

• GitHub: https://github.com/byExist/fluxgate
• Documentación: https://byExist.github.io/fluxgate/

¡Agradezco mucho sus comentarios!