Tips and Tricks #178: Implement Circuit Breaker for Resilient Services

Prevent cascade failures by implementing circuit breaker pattern for external service calls.

Code Snippet

import asyncio
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=timedelta(seconds=30)):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit is OPEN")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        self.failures = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = datetime.now()
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

Why This Helps

• Prevents cascade failures across services
• Allows failing services time to recover
• Provides fast failure instead of timeouts

How to Test

• Simulate downstream service failure
• Verify circuit opens after threshold

When to Use

Any external service call. APIs, databases, third-party services.

Performance/Security Notes

Consider using libraries like resilience4j (Java) or Polly (.NET) for production use.

References

• https://martinfowler.com/bliki/CircuitBreaker.html

Try this tip in your next project and share your results in the comments!