pybreaker_circuit_breaker_decorator_quickstart.py

1import pybreaker
2
3# Create a circuit breaker that will open after 5 consecutive failures
4# and will attempt to close again after 60 seconds.
5db_breaker = pybreaker.CircuitBreaker(
6    fail_max=5,
7    reset_timeout=60
8)
9
10@db_breaker
11def update_customer_data(customer_id, data):
12    """
13    This function will be protected by the circuit breaker.
14    If it raises exceptions repeatedly, the breaker will open.
15    """
16    # Imagine a database call here
17    pass
18
19# To call the function, just call it normally.
20# If the breaker is open, it will raise a pybreaker.CircuitBreakerError.
21try:
22    update_customer_data(42, {"name": "John Doe"})
23except pybreaker.CircuitBreakerError:
24    # Handle the case where the circuit is open
25    print("Circuit is open, request failed immediately")