pybreaker_circuit_breaker_decorator_with_failure_threshold.py

1import pybreaker
2import requests
3
4# Create a circuit breaker that will open after 5 consecutive failures
5# and will stay open for 60 seconds.
6db_breaker = pybreaker.CircuitBreaker(
7    fail_max=5,
8    reset_timeout=60
9)
10
11@db_breaker
12def update_customer_data(customer_id, data):
13    # This function is now protected by the circuit breaker.
14    # If it raises an exception, the breaker will count it as a failure.
15    # If the failure threshold is reached, the breaker opens and 
16    # subsequent calls will immediately raise pybreaker.CircuitBreakerError.
17    response = requests.post(f"https://api.example.com/customers/{customer_id}", json=data)
18    response.raise_for_status()
19    return response.json()
20
21# You can also use it manually or as a context manager
22def manual_example():
23    try:
24        with db_breaker:
25            # Code here is protected
26            pass
27    except pybreaker.CircuitBreakerError:
28        # The circuit is open
29        print("Circuit is open, skipping request.")