How to Compose Parallel API Calls in Python with asyncio.gather
Compose multiple mock API responses in parallel using asyncio.gather with per-service simulated latency.
Python code
23 linesimport asyncio
import random
import time
async def mock_api(name: str, delay: float) -> dict:
await asyncio.sleep(delay)
return {"service": name, "value": random.randint(1, 100)}
async def fetch_all():
services = {
"users": mock_api("users", 0.2),
"orders": mock_api("orders", 0.3),
"products": mock_api("products", 0.1),
}
start = time.perf_counter()
results = await asyncio.gather(*services.values(), return_exceptions=True)
elapsed = time.perf_counter() - start
combined = {name: result for name, result in zip(services.keys(), results)}
print(f"Total time: {elapsed:.2f}s")
print(f"Composed result: {combined}")
if __name__ == "__main__":
asyncio.run(fetch_all())
Output
Total time: 0.31s
Composed result: {'users': {'service': 'users', 'value': 42}, 'orders': {'service': 'orders', 'value': 87}, 'products': {'service': 'products', 'value': 15}}
How it works
The code creates three coroutines for mock services and schedules them all at once inside a dict. asyncio.gather runs them concurrently, so the wall-clock time equals the slowest request (~0.3s) rather than the sum of all delays. The return_exceptions=True flag prevents one failing service from cancelling the others. Because gather preserves input order, zipping the service keys with the results maps each response to its correct origin.
Common mistakes
- Calling the async function immediately in the dict, which schedules it; use `functools.partial` or lambdas if arguments differ per call.
- Forgetting `return_exceptions=True`, which makes a single API timeout crash the whole composition.
- Assuming `gather` returns results in completion order instead of input order.
Variations
- Use `asyncio.create_task` plus `asyncio.wait` for fine-grained control over timeouts and cancellations.
- Switch to `httpx.AsyncClient` with `.get` calls instead of mocks for real HTTP requests.
Real-world use cases
- A backend BFF (Backend-for-Frontend) aggregating user profiles, orders, and product data into one response.
- A payment orchestration service fetching customer, balance, and transaction history in parallel before decisioning.
- A metrics dashboard pulling health status from multiple microservices in a single refresh call.
Sponsored
More from Microservices patterns
- BFF aggregation pattern: combine multiple service responses in Python easy
- Backward Compatible Schema Evolution in Python medium
- Bulkhead Thread Pool per Service Mock in Python medium
- CQRS with Separate Read and Write Repositories in Python medium
- Cache-Aside Pattern in Python: Per-Service Mock easy
- Consumer Driven Contract Pact Mock in Python medium
Keep learning
Related tutorials and quizzes for this topic.