FastAPI Data Fetching with Async
Fetch and aggregate data from multiple APIs in FastAPI using asyncio.gather for concurrent requests, httpx connection pooling, response caching with TTL,.
Code
from fastapi import FastAPI
import httpx
import asyncio
app = FastAPI()
async def fetch_user(uid: int):
async with httpx.AsyncClient() as c:
r = await c.get(f'https://api.example.com/users/{uid}')
return r.json()
async def fetch_posts(uid: int):
async with httpx.AsyncClient() as c:
r = await c.get(f'https://api.example.com/users/{uid}/posts')
return r.json()
@app.get('/users/{uid}')
async def get_user_data(uid: int):
user, posts = await asyncio.gather(fetch_user(uid), fetch_posts(uid))
return {'user': user, 'posts': posts}