pybreaker_circuit_breaker_decorator_quickstart_example.py

1import pybreaker
2
3# Create a circuit breaker that will open after 5 consecutive failures
4# and will wait 60 seconds before attempting to close again.
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 is wrapped by the circuit breaker.
14    If it raises exceptions consistently, the breaker will open.
15    """
16    # Imagine a database operation here
17    pass
18
19# Alternatively, you can call functions manually
20try:
21    db_breaker.call(update_customer_data, 123, {"name": "John Doe"})
22except pybreaker.CircuitBreakerError:
23    # This exception is raised when the circuit is open
24    print("Circuit is open, request ignored")