Twenty Calls in the Time of One

Fan out IO-bound calls with one client and a semaphore, so twenty requests cost about one request's worth of waiting.

Syntaxasync with httpx.AsyncClient() as client: r = await client.get(url)

Twenty independent HTTP calls at 100ms each take two seconds in a loop and about 100ms concurrently, because that time was never yours to spend. The pattern has three parts every time: one client, a bound, a gather.

One client, many requests

Create the client once and pass it into every task. async with httpx.AsyncClient() as client: opens a connection pool; a fresh client per request throws away the pool and every keep-alive connection, and the TLS handshake often costs more than the request. Inside a task: r = await client.get(url), then r.raise_for_status() and r.json(), both synchronous in httpx.

The bound matters more than the speed

Unbounded fan-out is how you cause an outage — theirs, with 500 simultaneous requests, or yours, with exhausted sockets and a rate-limit ban. asyncio.Semaphore(n) caps how many tasks are inside the request at once, and the context manager is the shorthand for acquire/release:

async with semaphore:              # shorthand
    return await client.get(url)

await semaphore.acquire()          # the same thing, longhand
try:
    return await client.get(url)
finally:
    semaphore.release()

Choose n from the limit the other side publishes, not from your core count — concurrency here is not about CPUs. httpx has its own pool ceiling via httpx.Limits, which queues rather than fails; the semaphore is where your bound lives, visible and tunable. Then await asyncio.gather(*(fetch(client, u) for u in urls)) brings the results back in the order you asked for them.

Example

Try it yourself
Loading editor…
Press Run to execute the code.

When to use it

  • Hydrating 1,000 order ids from a vendor API inside a vendor rate limit.
  • A webhook dispatcher that must deliver to hundreds of customer endpoints without flooding any one of them.
  • Checking link health across a site and reporting which URLs failed.
  • Replacing a requests loop in a nightly import that now runs past its window.

More examples

The whole pattern, with httpx

The production shape of the runnable example above: client, semaphore sized to the remote limit, gather that tolerates individual failures. Matching httpx.Limits to the semaphore keeps the two ceilings from disagreeing, which otherwise shows up as requests mysteriously queueing inside the client.

Example · python
# fanout.py
import asyncio

import httpx

CONCURRENCY = 8          # from the vendor's published rate limit, not os.cpu_count()

async def fetch(client, sem, url):
    async with sem:                               # the bound, inside the task
        r = await client.get(url, timeout=10.0)
        r.raise_for_status()
        return r.json()

async def fetch_all(urls):
    sem = asyncio.Semaphore(CONCURRENCY)
    limits = httpx.Limits(max_connections=CONCURRENCY)
    # one client for the whole run: one pool, keep-alive connections reused
    async with httpx.AsyncClient(limits=limits) as client:
        return await asyncio.gather(
            *(fetch(client, sem, u) for u in urls),
            return_exceptions=True,               # report bad URLs, finish the batch
        )

if __name__ == "__main__":
    results = asyncio.run(fetch_all([f"https://api.example.com/items/{n}" for n in range(100)]))
    print(sum(1 for r in results if not isinstance(r, BaseException)), "of", len(results), "ok")

Pages in sequence, details in parallel

Most real API work is this hybrid: the listing has to be walked in order because each response carries the next cursor, but the per-item lookups are independent and can all be in flight at once. Collect the ids first, then fan out.

Example · python
import asyncio

import httpx

async def page_ids(client, url):
    """Pagination is inherently serial: you need page N to know page N+1."""
    while url:
        r = await client.get(url)
        body = r.json()
        for item in body["data"]:
            yield item["id"]
        url = body.get("next")          # cursor from the response

async def detail(client, sem, item_id):
    async with sem:
        return (await client.get(f"/items/{item_id}")).json()

async def main():
    sem = asyncio.Semaphore(10)
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        ids = [i async for i in page_ids(client, "/items")]   # serial walk
        details = await asyncio.gather(*(detail(client, sem, i) for i in ids))
    print(len(details), "items hydrated")

A separate bound per host

A single global semaphore means one unresponsive host can occupy every slot while the other hosts idle. Keying the semaphores by netloc fixes that. Reach for it in crawlers, webhook dispatchers and anything that fans out across domains you do not control.

Example · python
import asyncio
import collections
from urllib.parse import urlsplit

class PerHostLimiter:
    """One semaphore per hostname: a slow host cannot starve the others."""

    def __init__(self, per_host=4):
        self._per_host = per_host
        self._sems = collections.defaultdict(lambda: asyncio.Semaphore(per_host))

    def for_url(self, url):
        return self._sems[urlsplit(url).netloc]

async def fetch(client, limiter, url):
    async with limiter.for_url(url):        # bound applies to this host only
        return await client.get(url, timeout=10.0)

Related

py async gatherpy async timeoutspy web clientpy web paginationpy web retries

Discussion

  • Be the first to comment on this lesson.