Request Aggregation at the API Gateway
In this tutorial, you'll learn about Request Aggregation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Request aggregation combines multiple backend service calls into a single client response, reducing the number of round trips the client must make.
What You'll Learn
By the end of this lesson, you will implement request aggregation with parallel calls, response composition, error handling, and timeout management.
Why It Matters
Without aggregation, clients must make N+1 requests for related data. Aggregation reduces this to 1 request, improving latency and reducing client complexity.
Real-World Use
A dashboard page needs user info, recent orders, and notifications. The gateway calls three services in parallel and combines the results into one response.
Aggregation Flow
sequenceDiagram
Client->>Gateway: GET /dashboard
Gateway->>UserService: GET /users/me
Gateway->>OrderService: GET /orders/recent
Gateway->>NotifService: GET /notifications
UserService-->>Gateway: {user data}
OrderService-->>Gateway: {orders data}
NotifService-->>Gateway: {notifications}
Gateway-->>Client: {combined response}
Parallel Request Aggregation
# request_aggregation.py
import asyncio
import time
from typing import Any, Dict, List, Optional
class MockBackend:
async def fetch(self, service: str, endpoint: str, delay: float) -> Dict:
await asyncio.sleep(delay)
data = {
"users": {"id": 1, "name": "Alice", "email": "alice@example.com"},
"orders": [{"id": 101, "total": 29.99}],
"notifications": [{"text": "Order shipped"}],
"products": {"count": 42},
}
return data.get(service, {"error": "Unknown service"})
class RequestAggregator:
def __init__(self):
self.backend = MockBackend()
async def aggregate(self, requests: List[Dict], timeout: float = 5.0) -> Dict:
async def fetch_one(req: Dict) -> Dict:
return await self.backend.fetch(
req["service"], req["endpoint"], req.get("delay", 0.1)
)
try:
results = await asyncio.wait_for(
asyncio.gather(*[fetch_one(r) for r in requests]),
timeout=timeout,
)
except asyncio.TimeoutError:
return {"error": "Request aggregation timed out"}
combined = {}
for req, result in zip(requests, results):
combined[req["name"]] = result
return combined
aggregator = RequestAggregator()
async def main():
start = time.time()
result = await aggregator.aggregate([
{"service": "users", "endpoint": "/me", "name": "user", "delay": 0.2},
{"service": "orders", "endpoint": "/recent", "name": "orders", "delay": 0.3},
{"service": "notifications", "endpoint": "/unread", "name": "notifications", "delay": 0.1},
])
elapsed = time.time() - start
print(f"Combined response ({elapsed:.2f}s):")
for key, value in result.items():
print(f" {key}: {value}")
import asyncio
asyncio.run(main())
Expected output:
Combined response (0.30s):
user: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}
orders: [{'id': 101, 'total': 29.99}]
notifications: [{'text': 'Order shipped'}]
Aggregation with Error Handling
# aggregation_errors.py
import asyncio
from typing import Any, Dict, List, Optional
class ResilientAggregator:
async def aggregate(self, requests: List[Dict]) -> Dict:
async def fetch(req):
try:
mock = MockBackend()
result = await mock.fetch(req["service"], req["endpoint"], req.get("delay", 0.1))
return {"name": req["name"], "success": True, "data": result}
except Exception as e:
return {"name": req["name"], "success": False, "error": str(e)}
results = await asyncio.gather(*[fetch(r) for r in requests])
combined = {}
errors = []
for result in results:
if result["success"]:
combined[result["name"]] = result["data"]
else:
errors.append({result["name"]: result["error"]})
response = {"data": combined}
if errors:
response["errors"] = errors
return response
ra = ResilientAggregator()
async def main():
result = await ra.aggregate([
{"service": "users", "endpoint": "/me", "name": "user", "delay": 0.1},
{"service": "unknown", "endpoint": "/x", "name": "analytics", "delay": 0.1},
])
import json
print(json.dumps(result, indent=2))
asyncio.run(main())
Expected output:
{
"data": {
"user": {
"id": 1,
"name": "Alice",
"email": "alice@example.com"
}
},
"errors": [
{
"analytics": "Unknown service"
}
]
}
Common Mistakes
1. Sequential Instead of Parallel Calls
Calling services sequentially adds their latencies together. Use async requests to call them in parallel.
2. No Timeout for Individual Calls
One slow service delays the entire aggregation. Set per-service timeouts and continue without failed services.
3. Tight Coupling to Backend Schemas
If the user service changes its response format, the aggregation breaks. Define aggregation schemas separately.
4. Ignoring Partial Failures
When one service fails, the aggregation should still return data from successful services with error information.
5. Too Much Data in One Response
Aggregating too many services creates large response payloads. Limit aggregation to 3-5 related services per endpoint.
Practice Questions
1. Why is parallel execution important for aggregation?
Sequential calls sum latencies. Parallel calls complete in the time of the slowest service, reducing total latency.
2. How do you handle partial failures in aggregation?
Return successfully fetched data alongside error information. Let the client decide how to handle partial data.
3. What is the N+1 Problem in API requests?
The client makes 1 request for a list and N requests for each item's details. Aggregation solves this by batching related data.
4. How does aggregation affect Caching?
Aggregated responses are harder to cache because they combine data from multiple sources. Cache individual service responses instead.
Challenge
Design an aggregation endpoint for a product detail page that fetches product info, inventory status, pricing, reviews, and seller info in parallel, with timeouts and partial failure handling.
FAQ
Mini Project: Aggregation Gateway
# aggregation_gateway.py
import asyncio
import time
from typing import Any, Dict, List, Optional
class AggregationGateway:
def __init__(self):
self.services = {}
def register(self, name: str, fetch_fn, timeout: float = 2.0):
self.services[name] = {"fetch": fetch_fn, "timeout": timeout}
async def get_dashboard(self, required: List[str]) -> Dict:
async def fetch_service(name):
svc = self.services.get(name)
if not svc:
return {"name": name, "success": False, "error": "Unknown service"}
try:
data = await asyncio.wait_for(svc["fetch"](), timeout=svc["timeout"])
return {"name": name, "success": True, "data": data}
except asyncio.TimeoutError:
return {"name": name, "success": False, "error": "timeout"}
except Exception as e:
return {"name": name, "success": False, "error": str(e)}
results = await asyncio.gather(*[fetch_service(s) for s in required])
data = {}
errors = []
for r in results:
if r["success"]:
data[r["name"]] = r["data"]
else:
errors.append({r["name"]: r["error"]})
response = {"data": data}
if errors:
response["errors"] = errors
return response
gw = AggregationGateway()
async def fake_user():
await asyncio.sleep(0.1)
return {"id": 1, "name": "Alice"}
async def fake_orders():
await asyncio.sleep(0.2)
return [{"id": 101, "total": 49.99}]
gw.register("user", fake_user, 1.0)
gw.register("orders", fake_orders, 1.0)
async def main():
result = await gw.get_dashboard(["user", "orders"])
import json
print(json.dumps(result, indent=2))
asyncio.run(main())
Expected output:
{
"data": {
"user": {
"id": 1,
"name": "Alice"
},
"orders": [
{
"id": 101,
"total": 49.99
}
]
}
}
What's Next
You understand request aggregation. Next, learn about API versioning at the gateway, then explore WebSocket gateway.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro