Fluxgate - スライディングウィンドウと組み合わせ可能なルールをサポートする Python Circuit Breaker

既存の Python Circuit Breaker ライブラリ（circuitbreaker、pybreaker など）を使っていて、いくつか物足りない点がありました。

• 連続失敗カウントしか使えない: 一度成功するとリセットされるため、不安定なサービスを検知しにくい
• 復旧時に単一のテスト呼び出ししか許可されない: キャッシュウォームアップなどには不十分

Django Permission の演算子サポート、resilience4j のウィンドウ管理から着想を得ました。

インストール

pip install fluxgate  

基本的な使い方

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

デフォルト値: 直近 100 回の呼び出しのうち 50% が失敗するとトリップし、60 秒後に復旧を試行

非同期対応

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

主な機能

スライディングウィンドウ + 失敗率ベースのトリップ

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

→ 5 回連続で失敗、または直近 100 回の呼び出しのうち 50% が失敗するとトリップ

段階的な復旧（RampUp）

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

→ 復旧時に 10% から始めて 100% まで 60 秒かけて段階的にトラフィックを増加

既存ライブラリとの機能比較

機能	fluxgate	circuitbreaker	pybreaker	aiobreaker
スライディングウィンドウ	O	X	X	X
失敗率ベースのトリップ	O	X	X	X
組み合わせ可能なルール	O	X	X	X
段階的な復旧	O	X	X	X

リンク

• GitHub: https://github.com/byExist/fluxgate
• ドキュメント: https://byExist.github.io/fluxgate/

ぜひ多くのフィードバックをお寄せください。