Python Async in Practice: Call Many APIs without Waiting in Line

The starting point

The example script reads a list of 400 course URLs and checks each one: it requests the page, records the status code and response time, and writes a report. Done sequentially with a synchronous HTTP client, the run takes several minutes because each request waits for the previous one. The network is idle most of that time, which makes this a good fit for asyncio.

Before rewriting, record the current run time and save the report. You will compare both after the change.

Choose async-capable tools

asyncio only helps if every slow operation can be awaited. Use an async HTTP client such as httpx's AsyncClient or aiohttp, and create one client for the whole run so connections are reused. If part of the program must use a blocking library, call it through asyncio.to_thread so it does not freeze the event loop.

Write one coroutine for one item

Keep the per-item logic in a single coroutine that returns a result object rather than raising. That makes it easy to collect successes and failures together.

async def check(client, url, limit):
    async with limit:
        started = time.perf_counter()
        try:
            r = await client.get(url, timeout=10)
            return Result(url, r.status_code, time.perf_counter() - started, None)
        except httpx.HTTPError as error:
            return Result(url, None, time.perf_counter() - started, repr(error))

The limit argument is an asyncio.Semaphore shared by all tasks. It caps how many requests are in flight at once.

Run the items concurrently with a limit

Create the semaphore, the client and the tasks, then wait for all results. On Python 3.11 and later, asyncio.TaskGroup manages the tasks and cancels the rest if one raises unexpectedly; because the coroutine above returns errors as results, the group completes normally.

async def main(urls):
    limit = asyncio.Semaphore(20)
    async with httpx.AsyncClient(follow_redirects=True) as client:
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(check(client, u, limit)) for u in urls]
    return [t.result() for t in tasks]

results = asyncio.run(main(urls))

Choose the concurrency limit from the target service's rate limits and your own capacity. Twenty is a reasonable start for public pages; an API with a documented limit should be respected exactly.

Add timeouts and careful retries

Every request needs a timeout so one slow server cannot hold a slot forever. Retry only errors that may succeed later, such as connection resets, timeouts and HTTP 429 or 503 responses, and wait between attempts with exponential backoff plus a little random jitter so retries do not arrive in synchronised waves. Do not retry 404 or other errors that will fail the same way again.

Report results usefully

Write the report in the original order, then add a summary: total checked, failures by type and the slowest responses. Save failures with their error text so someone can act on them. Compare the new report with the sequential one for the same input; apart from timing, the results should match.

Know when not to use asyncio

• CPU-heavy steps, such as parsing large documents, do not get faster with asyncio; use a process pool.
• Small scripts with a few calls rarely justify the extra complexity.
• Libraries without async support block the loop unless wrapped with to_thread.
• Shared mutable state between coroutines still needs care, even on one thread, when it spans an await.

Checklist

• Work is I/O-bound and libraries are async-capable
• One client is reused for all requests
• Concurrency is limited with a semaphore
• Every request has a timeout
• Only retryable errors are retried, with backoff and jitter
• Results and failures are reported in a usable summary