Back to All Nuggets
Python python ⏱️ 2 min read
Production Implementation
Ask endoAI
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