Fluxgate - sliding window और संयोज्य नियमों को सपोर्ट करने वाला Python Circuit Breaker

मौजूदा Python Circuit Breaker लाइब्रेरीज़ (circuitbreaker, pybreaker आदि) का इस्तेमाल करते हुए कुछ कमियाँ महसूस हुईं।

• केवल लगातार failure count का उपयोग: एक बार success होते ही reset हो जाता है, इसलिए अस्थिर service का पता लगाना मुश्किल
• recovery के समय केवल एक test call की अनुमति: cache warming आदि के लिए पर्याप्त नहीं

Django Permission operator support और resilience4j के window management से प्रेरणा मिली।

इंस्टॉलेशन

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 calls में 50% failure होने पर trip, 60 सेकंड बाद recovery की कोशिश

async सपोर्ट

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

मुख्य फीचर्स

sliding window + failure rate आधारित trip

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

→ 5 बार लगातार failure होने पर, या हाल की 100 calls में 50% failure होने पर trip

क्रमिक recovery (RampUp)

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

→ recovery के समय 10% से शुरू होकर 60 सेकंड में ट्रैफ़िक धीरे-धीरे 100% तक बढ़ता है

मौजूदा लाइब्रेरीज़ के साथ फीचर तुलना

फीचर	fluxgate	circuitbreaker	pybreaker	aiobreaker
sliding window	O	X	X	X
failure rate आधारित trip	O	X	X	X
संयोज्य नियम	O	X	X	X
क्रमिक recovery	O	X	X	X

लिंक

• GitHub: https://github.com/byExist/fluxgate
• दस्तावेज़: https://byExist.github.io/fluxgate/

कृपया खूब फ़ीडबैक दें!