Implementing Circuit Breakers in Python

1. The Danger of Cascading Failures

In a monolithic application, a failure is often total—the entire application crashes. In a microservices architecture, the danger is more insidious: a failure in one small part of the system can trigger a chain reaction, leading to a cascading failure that brings down many other services.

Imagine our FastAPI gateway makes a gRPC call to the recommendations service. What happens if the recommendations service becomes very slow, perhaps due to a database lock or high CPU usage?

1. A request arrives at the gateway. The gateway calls the recommendations service.
2. The gateway's worker process (or thread) handling this request now blocks, waiting for a response.
3. More requests arrive at the gateway. More workers become tied up, all waiting on the slow recommendations service.
4. Eventually, the gateway exhausts its pool of available workers. It can no longer handle any incoming requests, even those for different, healthy services.
5. The gateway is now effectively down. The failure in the recommendations service has cascaded and taken the gateway with it.

This is not just a theoretical problem; it's a common cause of major outages in distributed systems. The following video provides an excellent visualization of how this resource exhaustion occurs.

2. The Circuit Breaker Pattern: Failing Fast

To prevent cascading failures, we need a mechanism that stops a client from repeatedly calling a service it knows is failing. This is where the circuit breaker pattern comes in. Much like an electrical circuit breaker in your home trips to prevent an overload from damaging your appliances, a software circuit breaker "trips" to protect an application from a failing dependency.

It acts as a stateful proxy that wraps the dangerous operation (like a network call) and monitors it for failures. It has three states:

1. 🟢 CLOSED: This is the normal state. Requests are allowed to pass through to the downstream service. The circuit breaker monitors the number of failures. If the failure count exceeds a configured threshold within a certain time period, the breaker "trips" and transitions to the OPEN state.

2. 🔴 OPEN: In this state, the circuit is "open." The breaker immediately rejects all requests for the downstream service without even attempting the network call. This is called failing fast. It prevents the calling service from tying up its resources and gives the failing downstream service time to recover. After a configured timeout period, the breaker transitions to the HALF-OPEN state.

3. 🟡 HALF-OPEN: The breaker allows a limited number of "trial" requests to pass through to the downstream service.

   • If these trial requests succeed, the breaker assumes the service has recovered and transitions back to the CLOSED state, restoring normal operation.
   • If any trial request fails, the breaker trips again and returns to the OPEN state, restarting the recovery timeout.

The following video provides a great walkthrough of these three states and how they work together to protect the system.

It's important to distinguish the circuit breaker pattern from the simple retry pattern we discussed in a previous lesson.

• Retries are best for handling transient (temporary) faults, like a brief network blip.
• Circuit Breakers are for handling more persistent faults, like a service being completely offline or severely degraded. In fact, a naive retry mechanism can make a cascading failure worse by bombarding an already-struggling service with even more requests.

3. Implementing a Circuit Breaker in Async Python

While you could implement a circuit breaker from scratch, several excellent libraries do the heavy lifting for you. For our async Python stack, the aiobreaker library is a perfect fit.

Let's walk through how to protect a gRPC client call using aiobreaker.

Step 1: Installation

First, add the library to your project's dependencies and install it.

pip install aiobreaker

Step 2: Implementing the Circuit Breaker

The core idea is to wrap your async function that makes the network call with the circuit breaker.

The article "Building Resilient Database Operations with Async SQLAlchemy" provides a clear, practical example of creating a circuit breaker class in Python. While the article's focus is on databases, the pattern is identical for protecting any external call, including our gRPC service.

Inspired by that example, let's create a circuit breaker for our gRPC client. Imagine you have a client function call_recommendation_service that makes the actual gRPC call.

import asyncio
import grpc
from aiobreaker import CircuitBreaker, CircuitBreakerError
from datetime import timedelta




# Assume these are your generated gRPC stubs
# import recommendations_pb2
# import recommendations_pb2_grpc




# 1. Configure the circuit breaker
# - Trips after 3 consecutive failures
# - Stays open for 30 seconds before moving to half-open
reco_service_breaker = CircuitBreaker(
    fail_max=3,
    timeout_duration=timedelta(seconds=30)
)




# This is our function that makes the potentially failing network call
async def call_recommendation_service(user_id: str):
    print("Attempting to call Recommendations service...")
    try:



        # Connect to the gRPC service
        async with grpc.aio.insecure_channel('recommendations:50051') as channel:



            # stub = recommendations_pb2_grpc.RecommendationsStub(channel)
            # request = recommendations_pb2.RecommendationRequest(user_id=user_id)
            # response = await stub.Recommend(request)
            



            # For demonstration, let's simulate a failure
            raise grpc.aio.AioRpcError(grpc.StatusCode.UNAVAILABLE, "Service is down")
            



            # return response
    except grpc.aio.AioRpcError as e:
        print(f"gRPC call failed: {e.details()}")



        # Re-raising the exception ensures the circuit breaker counts it as a failure
        raise e




# 2. Create a protected wrapper function
async def fetch_recommendations_protected(user_id: str):
    try:



        # 3. Wrap the call using the breaker
        response = await reco_service_breaker.call_async(call_recommendation_service, user_id)
        print("Success: Fetched recommendations.")
        return {"data": response, "status": "ok"}
    except CircuitBreakerError:



        # 4. Handle the open circuit: Fail fast!
        print("Circuit is open. Failing fast and returning fallback.")
        return {"data": [], "status": "fallback", "error": "Recommendations service is temporarily unavailable."}
    except grpc.aio.AioRpcError:



        # Handle the initial failures before the circuit opens
        print("gRPC call failed. Returning fallback.")
        return {"data": [], "status": "fallback", "error": "Could not connect to recommendations service."}




# Example of how to use it
async def main():



    # The first 3 calls will fail and be counted by the breaker
    for i in range(4):
        print(f"\n--- Request {i+1} ---")
        await fetch_recommendations_protected("user-123")
        await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Dissecting the code:

1. Configuration (CircuitBreaker): We create an instance of CircuitBreaker, setting fail_max to the number of failures that will trip the circuit and timeout_duration to the time it will stay open.
2. Protected Function (fetch_recommendations_protected): This function acts as the public interface for your client. It contains the core try...except logic.
3. Wrapping the Call (breaker.call_async): Instead of calling call_recommendation_service directly, we pass it to reco_service_breaker.call_async. The breaker executes our function and monitors its outcome. If it raises an exception (that isn't in the exclude list), the breaker increments its failure counter.
4. Handling CircuitBreakerError: If you call this function while the breaker is in the OPEN state, call_async doesn't even run your function. It immediately raises CircuitBreakerError. Your except block for this error is your fallback logic. This is where you would return a cached response, a default value, or raise a 503 Service Unavailable error to the end-user.

4. Hands-on Exercise

Now it's your turn to apply this pattern. Below is a simple async function that simulates fetching a user's profile from an external service, which fails intermittently.

Your task is to protect this call with a circuit breaker.

Instructions:

1. Add the aiobreaker library.
2. Create a CircuitBreaker instance that trips after 2 failures and stays open for 5 seconds.
3. Modify the get_user_profile function to use the circuit breaker.
4. Implement a try...except CircuitBreakerError block to return a fallback dictionary {"status": "fallback", "name": "Guest"} when the circuit is open.
5. Run the code and observe how it behaves: it should try twice, fail, open the circuit, fail fast immediately, wait 5 seconds, and then try again.

# Starter code for the exercise
import asyncio
import random




# --- Your code goes here ---




# A mock function that simulates a failing network call
call_counter = 0
async def fetch_from_remote_service(user_id: int):
    global call_counter
    call_counter += 1
    print(f"Attempting to fetch profile for user {user_id}... (Call #{call_counter})")
    



    # Simulate a persistent failure for the first few calls
    if call_counter < 4:
        await asyncio.sleep(0.5) # Simulate network latency
        raise ConnectionError("Service is unavailable")
    
    print("Success! Service has recovered.")
    return {"name": "Sagarmatha", "id": user_id}




# This is the main function that needs protection
async def get_user_profile(user_id: int):
    try:



        # TODO: Protect the call to fetch_from_remote_service
        profile = await fetch_from_remote_service(user_id)
        return profile
    except ConnectionError:
        print("Caught a connection error.")
        return {"status": "error", "name": "Unknown"}

async def main():
    print("--- Simulating service outage ---")



    # Calls during the outage
    for i in range(3):
        profile = await get_user_profile(101)
        print(f"Received: {profile}\n")
        await asyncio.sleep(1)

    print("\n--- Waiting for circuit to enter half-open state ---\n")
    await asyncio.sleep(5)

    print("\n--- Simulating service recovery ---")



    # This call should be in the half-open state
    profile = await get_user_profile(101)
    print(f"Received: {profile}\n")
    



    # If the previous call succeeded, the circuit should now be closed
    profile = await get_user_profile(101)
    print(f"Received: {profile}\n")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import random
from aiobreaker import CircuitBreaker, CircuitBreakerError
from datetime import timedelta




# 1. Add aiobreaker and create a CircuitBreaker instance
profile_breaker = CircuitBreaker(fail_max=2, timeout_duration=timedelta(seconds=5))




# A mock function that simulates a failing network call
call_counter = 0
async def fetch_from_remote_service(user_id: int):
    global call_counter
    call_counter += 1
    print(f"Attempting to fetch profile for user {user_id}... (Call #{call_counter})")
    



    # Simulate a persistent failure for the first few calls
    if call_counter < 4:
        await asyncio.sleep(0.5) # Simulate network latency
        raise ConnectionError("Service is unavailable")
    
    print("Success! Service has recovered.")
    return {"name": "Sagarmatha", "id": user_id}




# This is the main function that needs protection
async def get_user_profile(user_id: int):
    try:



        # 2. Protect the call to fetch_from_remote_service
        profile = await profile_breaker.call_async(fetch_from_remote_service, user_id)
        return profile
    except CircuitBreakerError:



        # 3. Handle the open circuit case
        print("Circuit is open! Returning fallback.")
        return {"status": "fallback", "name": "Guest"}
    except ConnectionError:
        print("Caught a connection error.")
        return {"status": "error", "name": "Unknown"}

async def main():
    print("--- Simulating service outage ---")



    # Calls during the outage
    for i in range(3):
        profile = await get_user_profile(101)
        print(f"Received: {profile}\n")
        await asyncio.sleep(1)

    print("\n--- Waiting for circuit to enter half-open state ---\n")
    await asyncio.sleep(5)

    print("\n--- Simulating service recovery ---")



    # This call should be in the half-open state
    profile = await get_user_profile(101)
    print(f"Received: {profile}\n")
    



    # If the previous call succeeded, the circuit should now be closed
    profile = await get_user_profile(101)
    print(f"Received: {profile}\n")

if __name__ == "__main__":
    asyncio.run(main())

Conclusion

You've just implemented a powerful pattern for building self-healing, resilient systems. By preventing a client from hammering a failing service, the circuit breaker protects the client's own resources and gives the downstream service a chance to recover, dramatically reducing the likelihood of a cascading failure.

Key Takeaways:

• Cascading failures are a significant threat in microservice architectures, where a single slow service can exhaust resources in upstream services.
• The Circuit Breaker pattern mitigates this by acting as a stateful proxy that monitors and controls requests to a dependency.
• The three states—Closed, Open, and Half-Open—allow the system to "fail fast" and gracefully recover.
• In Python, async-native libraries like aiobreaker make it easy to wrap external calls and implement fallback logic by catching CircuitBreakerError.

Preview of the Next Lesson

We've made our client more resilient, but what about security? Currently, any service on our Docker network can call any other service. In the next lesson, we will begin to lock things down by implementing JWT-based authentication in our FastAPI API Gateway. You will learn how to issue tokens and protect endpoints, ensuring that only authenticated users can access your system.