Python asyncio for data ingestion: gather, semaphores and timeouts

Pulling data from an API is slow in a particular way: most of the time your program is not working, it is waiting for a reply. Python's asyncio lets one program wait for many replies at once. This guide uses a pretend API so every number is reproducible, and shows the four things you need for real ingestion: running calls together, keeping the order of results, limiting how many run at once, and handling a call that fails.

What you will learn

• What a coroutine is and how async and await work.
• How asyncio.gather runs many calls at the same time and keeps results in order.
• How to limit the number of calls running at once with a semaphore.
• How to handle one failing call without losing the others, and how to set a timeout.
• When asyncio does not help, so you do not use it for the wrong job.

1The problem: 5 pages, one at a time

An API returns 5 pages of orders. Each call takes 0.2 seconds, because most of that time is the network. We pretend with sleep, which waits without doing any work.

import asyncio
import time

PAGES = [1, 2, 3, 4, 5]

def fetch_page_blocking(page):
    time.sleep(0.2)
    return f"page {page}"

start = time.perf_counter()
results = [fetch_page_blocking(page) for page in PAGES]
elapsed = time.perf_counter() - start

print(results)
print("took at least 1.0 seconds:", elapsed >= 1.0)

['page 1', 'page 2', 'page 3', 'page 4', 'page 5']
took at least 1.0 seconds: True

5 calls of 0.2 seconds, one after another, take a full second. The program spent almost all of that second doing nothing. With 500 pages it would be 100 seconds.

2Coroutines: async and await

A function written with async def is a coroutine. Calling it does not run it; it gives you a coroutine object. await is where it can pause and let another coroutine run. asyncio.run starts the loop that switches between them.

async def fetch_page(page):
    await asyncio.sleep(0.2)
    return f"page {page}"

coro = fetch_page(1)
print(type(coro).__name__)
print(asyncio.run(coro))

coroutine
page 1

The first line shows that calling fetch_page did not run anything yet; it only created the coroutine. The second line is the result after asyncio.run executed it.

3asyncio.gather: run them together

Now start all 5 calls and wait for all of them. While one call waits, the loop starts the next, so the waits overlap.

async def fetch_all():
    return await asyncio.gather(*(fetch_page(page) for page in PAGES))

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

print(results)
print("took under 0.5 seconds:", elapsed < 0.5)

['page 1', 'page 2', 'page 3', 'page 4', 'page 5']
took under 0.5 seconds: True

The total fell from about 1 second to about 0.2 seconds, which is the time of the slowest single call.

The results keep the order you asked for

gather returns results in the order you passed the calls in, not the order they finished. Here 3 calls take different times.

finished = []

async def fetch_slow(name, delay):
    await asyncio.sleep(delay)
    finished.append(name)
    return name

async def three_calls():
    return await asyncio.gather(
        fetch_slow("a", 0.3), fetch_slow("b", 0.1), fetch_slow("c", 0.2)
    )

results = asyncio.run(three_calls())
print("results in request order:", results)
print("finished in this order:  ", finished)

results in request order: ['a', 'b', 'c']
finished in this order:   ['b', 'c', 'a']

Call a was started first and finished last. The results list still starts with a, so you can match each result to its request without extra work.

4Do not hit the API with everything at once

Starting 500 calls together can get you blocked, or run you out of connections. A semaphore is a counter that lets only a fixed number of coroutines into a section at once. Here at most 2 calls may run together, and we record the highest number that really did.

running = 0
max_running = 0
limit = asyncio.Semaphore(2)

async def fetch_limited(page):
    global running, max_running
    async with limit:
        running += 1
        max_running = max(max_running, running)
        await asyncio.sleep(0.1)
        running -= 1
        return f"page {page}"

async def fetch_all_limited():
    return await asyncio.gather(*(fetch_limited(page) for page in PAGES))

results = asyncio.run(fetch_all_limited())
print(results)
print("most calls running at once:", max_running)

['page 1', 'page 2', 'page 3', 'page 4', 'page 5']
most calls running at once: 2

All 5 pages came back, and never more than 2 were in flight. In real ingestion, choose the limit from what the API allows, and create the semaphore inside the loop that uses it, as these examples do.

5When one call fails

By default, if one call raises an error, gather raises it and you lose sight of the others. With return_exceptions=True every call finishes, and a failure shows up in the list as an exception object.

async def fetch_maybe_fail(page):
    await asyncio.sleep(0.05)
    if page == 3:
        raise ValueError("server error on page 3")
    return f"page {page}"

async def fetch_safely():
    return await asyncio.gather(*(fetch_maybe_fail(page) for page in PAGES), return_exceptions=True)

results = asyncio.run(fetch_safely())
for page, result in zip(PAGES, results):
    kind = "FAILED" if isinstance(result, Exception) else "ok"
    print(page, kind, result)

1 ok page 1
2 ok page 2
3 FAILED server error on page 3
4 ok page 4
5 ok page 5

Page 3 failed and the other four succeeded. Now your code can retry just page 3. A call that hangs forever is a different problem, and asyncio.wait_for solves it with a time limit.

async def hangs():
    await asyncio.sleep(10)

async def with_timeout():
    try:
        await asyncio.wait_for(hangs(), timeout=0.1)
    except asyncio.TimeoutError:
        return "gave up after 0.1 seconds"

print(asyncio.run(with_timeout()))

gave up after 0.1 seconds

6Threads as another option

Not every library supports asyncio. A popular blocking library, such as the requests package, cannot be awaited. For those, a thread pool gives the same overlap, because a thread that is waiting for the network lets the others run.

from concurrent.futures import ThreadPoolExecutor

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch_page_blocking, PAGES))
elapsed = time.perf_counter() - start

print(results)
print("took under 0.5 seconds:", elapsed < 0.5)

['page 1', 'page 2', 'page 3', 'page 4', 'page 5']
took under 0.5 seconds: True

The result and the speed-up are the same. Use asyncio when your HTTP library supports it, and threads when it does not.

7When asyncio does not help

Asyncio speeds up waiting. It does not speed up computing. If each page needs a long calculation, such as hashing every row or parsing a large file, the coroutines still run one at a time on one core, because only one can use the processor at any moment. For that kind of work, use multiple processes, or let Spark spread it over many machines.

A rule of thumb: if the job spends its time waiting for the network or the disk, asyncio or threads help. If it spends its time calculating, they do not.

# Real HTTP with asyncio needs a library that supports it, for example:
#   async with httpx.AsyncClient() as client:
#       responses = await asyncio.gather(*(client.get(url) for url in urls))
# The pattern is the same as the examples above.

8Check yourself

• 5 calls take 0.2 seconds each. About how long do they take one after another, and about how long with gather?
• Call a starts first but finishes last. Where does its result appear in the list gather returns?
• What does return_exceptions=True change about how a failure is reported?
• Why does asyncio not make a CPU-heavy hashing job faster?

Practise it: try Retry a flaky operation with a cap in the Python editor, and read the Concurrency, asyncio, and the GIL lesson in the Core Python track.