Show GN: Fluxgate - 슬라이딩 윈도우와 조합 가능한 규칙을 지원하는 Python Circuit Breaker

(byexist.github.io)

기존 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/

많은 피드백 부탁드립니다!