import asyncio
import httpx
from typing import List, Dict, Any
async def fetch_status(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, url: str) -> Dict[str, Any]:
async with semaphore:
try:
resp = await client.get(url, timeout=10.0)
return {'url': url, 'status_code': resp.status_code, 'healthy': resp.status_code == 200}
except Exception as e:
return {'url': url, 'status_code': None, 'healthy': False, 'error': str(e)}
async def check_all(urls: List[str]) -> List[Dict[str, Any]]:
sem = asyncio.Semaphore(15)
async with httpx.AsyncClient(limits=httpx.Limits(max_connections=50)) as client:
tasks = [fetch_status(client, sem, url) for url in urls]
return await asyncio.gather(*tasks)
Speeding Up Batch API Integrations with Asynchronous Concurrency
When fetching data from 500 external endpoints sequentially, execution takes hundreds of seconds. Using asyncio and httpx, requests run concurrently in parallel in seconds.
- Connection Pooling: Reuses TCP connections via
httpx.AsyncClient(). - Semaphore Rate Limiting: Use
asyncio.Semaphore(20)to avoid upstream throttling.
Accelerate Your Engineering Career
Get monthly masterclasses, exclusive code snippets, and direct architectural reviews.
Become a Supporter • KES 500 / Month