API Gateway Aggregation with gRPC

1. The Aggregator Pattern: Simplifying the Client

In a microservices architecture, data that constitutes a single conceptual resource for a user is often spread across multiple services. For example, a "Product Details" page might need information from a ProductInfoService, an InventoryService, a PricingService, and a ReviewService.

As the article explains, forcing the client (e.g., a mobile app) to make all these individual requests is inefficient and complex. The client would need to know the address of every service and orchestrate the calls itself. The Aggregator Pattern solves this by moving the orchestration logic to the API Gateway.

To understand the different strategies for implementing this pattern, let's watch a short video.

2. Building an Aggregating Gateway: A Restaurant Order

Let's build a practical example based on a scenario you requested: a restaurant kitchen. Imagine a client wants to place a complete order for a drink, a meal, and a dessert. In our microservices system, these items are managed by three separate services:

1. Bar Service (gRPC)
2. Kitchen Service (gRPC)
3. Bakery Service (gRPC)

Our FastAPI gateway will expose a single endpoint, POST /orders, that accepts the entire order. It will then concurrently call the three backend services to check the availability of each item and return a consolidated status to the client.

The following article provides a complete walkthrough of this exact scenario. We will use it as our guide for the implementation.

3. Implementation: Scatter-Gather with asyncio.gather

Now, let's focus on the core logic within our FastAPI gateway. The goal is to call the Bar, Kitchen, and Bakery services concurrently. If we were to call them sequentially (await call_bar(), await call_kitchen(), await call_bakery()), the total latency would be the sum of all three call latencies. By running them concurrently, the total latency will be determined by the longest of the three calls. This is the essence of the Scatter-Gather strategy.

In Python, asyncio.gather is the perfect tool for this job.

Let's assume we have client classes for each service, as shown in the article (BarClient, KitchenClient, BakeryClient). Each client has a method like get_order(item_name) that returns the status.

Here's how we would implement the aggregation logic in our FastAPI endpoint:

gateway_app/main.py

import asyncio
from fastapi import FastAPI, Depends
from uuid import UUID




# Import the Pydantic models and client classes from the article's structure
from schemas.orders import OrderCreate, OrderRead 
from api.dependencies.grpc.bar import BarClient
from api.dependencies.grpc.kitchen import KitchenClient
from api.dependencies.grpc.bakery import BakeryClient

app = FastAPI()




# --- Dependencies to provide gRPC clients ---
# In a real app, you might manage these as singletons.

def get_bar_client():
    return BarClient()

def get_kitchen_client():
    return KitchenClient()

def get_bakery_client():
    return BakeryClient()


@app.post("/orders", response_model=OrderRead, status_code=201)
async def create_order(
    order_create: OrderCreate,
    bar: BarClient = Depends(get_bar_client),
    kitchen: KitchenClient = Depends(get_kitchen_client),
    bakery: BakeryClient = Depends(get_bakery_client),
):
    """
    Receives an order, scatters requests to downstream services concurrently,
    gathers the results, and returns an aggregated response.
    """



    # 1. Scatter: Create coroutines for each gRPC call.
    #    The calls don't run yet; we are just scheduling them.
    drink_task = bar.get_order(order_create.drink)
    meal_task = kitchen.get_order(order_create.meal)
    dessert_task = bakery.get_order(order_create.dessert)




    # 2. Gather: Run all tasks concurrently and wait for all to complete.
    #    asyncio.gather collects the results in the same order.
    results = await asyncio.gather(
        drink_task,
        meal_task,
        dessert_task
    )
    



    # The client classes in the article handle gRPC errors and return a dict
    # like {"order_status": "details"}. Let's extract the status.
    drink_status = results[0].get("order_status", "Error")
    meal_status = results[1].get("order_status", "Error")
    dessert_status = results[2].get("order_status", "Error")




    # 3. Aggregate: Combine the results into a single response model.
    response = OrderRead(
        order_id=order_create.order_id,
        drink=f"{order_create.drink}: {drink_status}",
        meal=f"{order_create.meal}: {meal_status}",
        dessert=f"{order_create.dessert}: {dessert_status}",
    )

    return response

Code Breakdown:

1. Scatter: We create three awaitable objects (drink_task, meal_task, dessert_task). At this stage, the gRPC calls have not yet been executed. We have simply defined the concurrent work we want to do.
2. Gather: await asyncio.gather(...) is the key. It takes our awaitables and runs them concurrently on the asyncio event loop. The await pauses the create_order function until all three gRPC calls have completed. The results variable will be a list containing the return values from each coroutine, in the order they were passed to gather.
3. Aggregate: We process the results list, format the data, and construct our final OrderRead response model. The client receives a single, clean JSON object with the status of their entire order.

This implementation effectively hides the fact that three separate network calls were made to three different microservices.

4. Design Thinking: Handling Partial Failures

The asyncio.gather approach is highly efficient, but it has a specific behavior with exceptions. If any one of the awaited tasks raises an exception, gather will immediately propagate that exception, and the results of other, successfully completed tasks will be lost.

Question for you: In our restaurant order scenario, imagine the BakeryService is down and its gRPC call fails. With the default asyncio.gather behavior, the entire /orders request would fail. However, maybe the restaurant still wants to confirm the drink and meal part of the order.

How could you modify the asyncio.gather call to ensure that you can process results from the services that succeeded, even if one of them fails?

results = await asyncio.gather(
    drink_task,
    meal_task,
    dessert_task,
    return_exceptions=True  # This is the key change
)

responses = {}
for result in results:
    if isinstance(result, Exception):



        # Handle the error, e.g., log it and set a default status
        status = "Service Unavailable"



        # ... add logic to identify which item failed ...
    else:



        # Process the successful result
        status = result.get("order_status", "Error")



        # ...

Conclusion

In this lesson, you elevated your API Gateway from a simple proxy to a powerful orchestrator. You implemented the request aggregation pattern, a cornerstone of microservice architecture that simplifies client applications and improves efficiency.

Key Takeaways:

• The Aggregator Pattern moves the logic of calling multiple services from the client to the API Gateway.
• The Scatter-Gather strategy, implemented with asyncio.gather, is a highly performant way to make concurrent, independent downstream calls.
• Running calls concurrently minimizes latency, making the total response time dependent on the slowest service rather than the sum of all service response times.
• Properly designing an aggregator involves thinking about partial failures. Using features like asyncio.gather(..., return_exceptions=True) allows your gateway to be more resilient and provide a degraded but still useful response.

Preview of the Next Lesson

Our gateway is now functionally complete for handling client requests. But how do we manage it in a production environment? How does an orchestrator like Kubernetes know if our service is healthy and ready to receive traffic?

In the next lesson, we will address this by implementing health check (/healthz) and readiness (/readyz) endpoints. These are simple but critical API endpoints that provide vital signals to container orchestration systems, enabling automated recovery and zero-downtime deployments.