Fluxgate - Circuit Breaker em Python com suporte a regras combináveis e janela deslizante
(byexist.github.io)Ao usar bibliotecas existentes de Circuit Breaker para Python (circuitbreaker, pybreaker etc.), senti falta de alguns pontos.
- Usam apenas contagem de falhas consecutivas: um único sucesso reinicia tudo, dificultando detectar serviços instáveis
- Na recuperação, permitem apenas uma chamada de teste: insuficiente para casos como aquecimento de cache
A inspiração veio do suporte a operadores de permissão do Django e do gerenciamento de janelas do resilience4j.
Instalação
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")
Padrão: dispara com 50% de falhas nas últimas 100 chamadas e tenta se recuperar após 60 segundos
Suporte assí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")
Principais recursos
Janela deslizante + disparo baseado em taxa de falha
from fluxgate import CircuitBreaker
from fluxgate.trippers import MinRequests, FailureRate, FailureStreak
cb = CircuitBreaker(
name="api",
tripper=FailureStreak(5) | (MinRequests(100) & FailureRate(0.5))
)
→ Dispara com 5 falhas consecutivas ou 50% de falhas nas últimas 100 chamadas
Recuperação gradual (RampUp)
from fluxgate.permits import RampUp
cb = CircuitBreaker(
name="api",
permit=RampUp(initial=0.1, final=1.0, duration=60.0)
)
→ Na recuperação, começa em 10% e aumenta gradualmente o tráfego até 100% ao longo de 60 segundos
Comparação de recursos com bibliotecas existentes
| Recurso | fluxgate | circuitbreaker | pybreaker | aiobreaker |
|---|---|---|---|---|
| Janela deslizante | O | X | X | X |
| Disparo por taxa de falha | O | X | X | X |
| Regras combináveis | O | X | X | X |
| Recuperação gradual | O | X | X | X |
Links
- GitHub: https://github.com/byExist/fluxgate
- Documentação: https://byExist.github.io/fluxgate/
Agradeço muito qualquer feedback!
Ainda não há comentários.