Concurrent HTTP Requests with asyncio.gather and aiohttp

Fetching 100 endpoints one after another costs you 100 round-trips of latency. asyncio.gather fires them concurrently, so total wall time collapses to roughly the slowest request. What follows covers concurrent fetching with aiohttp: error handling, a semaphore for rate limiting, timeouts and connection pooling.

When to Use This

• Pulling data from several APIs in one go
• Web scraping with concurrent page downloads
• Batch processing of HTTP-based tasks (e.g., calling 100 LLM endpoints)
• Any I/O-bound workload where requests are independent

Prerequisites

• Python 3.10+
• aiohttp package (pip install aiohttp)

Solution

A consolidated runnable version lives in the companion repository.

1. Install Dependencies

pip install aiohttp

2. Basic Concurrent Fetch

import asyncio
import aiohttp
import time

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    """Fetch a single URL and return status + content.

    Args:
        session: aiohttp client session.
        url: URL to fetch.

    Returns:
        Dict with url, status, and text.
    """
    async with session.get(url) as response:
        return {
            "url": url,
            "status": response.status,
            "text": await response.text(),
        }

async def fetch_all(urls: list[str]) -> list[dict]:
    """Fetch all URLs concurrently.

    Args:
        urls: List of URLs to fetch.

    Returns:
        List of result dicts in the same order as input URLs.
    """
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Usage
urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

start = time.time()
results = asyncio.run(fetch_all(urls))
elapsed = time.time() - start

print(f"Fetched {len(urls)} URLs in {elapsed:.2f}s (concurrent)")
# ~1.2s instead of ~5s sequential

3. Error Handling with return_exceptions

async def fetch_all_safe(urls: list[str]) -> list[dict | Exception]:
    """Fetch all URLs, capturing exceptions instead of failing.

    Args:
        urls: List of URLs.

    Returns:
        List of results or Exception objects.
    """
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        # return_exceptions=True prevents one failure from canceling all
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# Process results, handling exceptions
results = asyncio.run(fetch_all_safe(urls))
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"URL {i} failed: {result}")
    else:
        print(f"URL {i}: status {result['status']}")

4. Concurrency Limiting with Semaphore

async def fetch_with_limit(
    urls: list[str],
    max_concurrent: int = 10,
) -> list[dict | Exception]:
    """Fetch URLs with a concurrency limit to avoid overwhelming servers.

    Args:
        urls: List of URLs.
        max_concurrent: Maximum simultaneous requests.

    Returns:
        List of results or exceptions.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(session: aiohttp.ClientSession, url: str) -> dict:
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Limit to 10 concurrent requests
results = asyncio.run(fetch_with_limit(urls, max_concurrent=10))

5. Timeouts

import aiohttp

async def fetch_with_timeout(
    session: aiohttp.ClientSession,
    url: str,
    timeout_seconds: float = 10.0,
) -> dict:
    """Fetch with a per-request timeout.

    Args:
        session: aiohttp session.
        url: URL to fetch.
        timeout_seconds: Timeout in seconds.

    Returns:
        Result dict or timeout error.
    """
    timeout = aiohttp.ClientTimeout(total=timeout_seconds)
    try:
        async with session.get(url, timeout=timeout) as response:
            return {
                "url": url,
                "status": response.status,
                "text": await response.text(),
            }
    except asyncio.TimeoutError:
        return {"url": url, "status": 0, "error": "timeout"}
    except aiohttp.ClientError as e:
        return {"url": url, "status": 0, "error": str(e)}

async def fetch_all_with_timeouts(urls: list[str], timeout: float = 10.0) -> list[dict]:
    """Fetch all URLs with timeouts."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_timeout(session, url, timeout) for url in urls]
        return await asyncio.gather(*tasks)

6. Connection Pooling with Session Config

async def fetch_with_pool(
    urls: list[str],
    max_concurrent: int = 20,
) -> list[dict | Exception]:
    """Fetch with optimized connection pool settings.

    Args:
        urls: List of URLs.
        max_concurrent: Max concurrent requests.

    Returns:
        List of results.
    """
    # Configure connection pool
    connector = aiohttp.TCPConnector(
        limit=max_concurrent,       # Total connection limit
        limit_per_host=5,           # Per-host limit
        ttl_dns_cache=300,          # DNS cache TTL in seconds
        enable_cleanup_closed=True,
    )

    timeout = aiohttp.ClientTimeout(
        total=30,      # Total timeout
        connect=10,    # Connection timeout
        sock_read=10,  # Socket read timeout
    )

    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

7. Progress Tracking

async def fetch_with_progress(urls: list[str]) -> list[dict | Exception]:
    """Fetch URLs with real-time progress tracking."""
    results = [None] * len(urls)
    completed = 0

    async def fetch_and_track(session: aiohttp.ClientSession, index: int, url: str):
        nonlocal completed
        try:
            result = await fetch_url(session, url)
            results[index] = result
        except Exception as e:
            results[index] = e
        finally:
            completed += 1
            print(f"\rProgress: {completed}/{len(urls)}", end="", flush=True)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_and_track(session, i, url) for i, url in enumerate(urls)]
        await asyncio.gather(*tasks)

    print()  # New line after progress
    return results

How It Works

1. asyncio.gather(*tasks) gets every coroutine running at once. The event loop interleaves them, suspending a task on each await and resuming it when the I/O completes.
2. aiohttp.ClientSession is where the connection pool lives. Keep a single session for all requests and its TCP connections get recycled, instead of paying a handshake on every call.
3. asyncio.Semaphore caps in-flight operations. Once max_concurrent tasks hold slots, the rest queue until one is released.
4. return_exceptions=True turns failures into values in the result list instead of raising, so one dead endpoint doesn’t cancel the other 99 in-flight requests.
5. ClientTimeout sets per-request deadlines: total bounds the whole exchange, connect the TCP handshake, sock_read the response read.

Variants

Batch Processing with Chunks

async def fetch_in_batches(
    urls: list[str],
    batch_size: int = 50,
) -> list[dict | Exception]:
    """Fetch URLs in batches to control memory and rate."""
    results = []

    for i in range(0, len(urls), batch_size):
        batch = urls[i:i + batch_size]
        batch_results = await fetch_all_safe(batch)
        results.extend(batch_results)
        print(f"Completed batch {i // batch_size + 1}")

    return results

Retry with Exponential Backoff

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> dict:
    """Fetch with retry and exponential backoff."""
    for attempt in range(max_retries):
        try:
            async with session.get(url) as response:
                if response.status == 429:
                    raise aiohttp.ClientError("Rate limited")
                return {
                    "url": url,
                    "status": response.status,
                    "text": await response.text(),
                }
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                return {"url": url, "error": str(e), "attempts": attempt + 1}
            delay = base_delay * (2 ** attempt)
            print(f"Retry {attempt + 1}/{max_retries} for {url} in {delay}s")
            await asyncio.sleep(delay)

async def fetch_all_with_retry(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_retry(session, url) for url in urls]
        return await asyncio.gather(*tasks)

POST Requests with JSON

async def post_json(
    session: aiohttp.ClientSession,
    url: str,
    data: dict,
) -> dict:
    """Send a POST request with JSON body."""
    async with session.post(url, json=data) as response:
        return {
            "url": url,
            "status": response.status,
            "json": await response.json(),
        }

async def post_all(
    url: str,
    payloads: list[dict],
) -> list[dict]:
    """Send multiple POST requests concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [post_json(session, url, payload) for payload in payloads]
        return await asyncio.gather(*tasks)

Using httpx (Alternative to aiohttp)

import httpx
import asyncio

async def fetch_httpx(urls: list[str]) -> list[dict]:
    """Concurrent fetch using httpx (syncs with requests API)."""
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [
            {"url": str(r.url), "status": r.status_code, "text": r.text}
            for r in responses
        ]

Best Practices

• Reuse ClientSession: a fresh session per request throws away connection pooling
• Set limit_per_host: keeps you from hammering a single server with everything at once
• Bound concurrency with a semaphore: respect the API’s rate limit, not just your throughput. The semaphore rate-limiting recipe goes deeper
• Always set timeouts: without them, one hanging server can stall the entire application

Common Mistakes

• Creating a new ClientSession per request: defeats pooling; create the session once and share it
• Forgetting return_exceptions=True: the first failed request cancels every other in-flight one
• No concurrency limit: firing 10,000 requests at once overwhelms your machine and the server
• Calling asyncio.run() inside a running event loop (Jupyter, FastAPI): raises RuntimeError; await the coroutine directly instead

See Also

• asyncio.gather documentation: official semantics, including how return_exceptions works.
• aiohttp client reference: everything about sessions, connectors and timeouts.
• httpx async guide: the alternative client used in the variants section.
• Python Asyncio in Production: the deeper guide on supervision, shutdown and observability.