Lesson 1: Async Fundamentals

The problem this lesson solves

The recipe-extractor pipeline processes documents one at a time. Each document needs one embedding call and one extraction call, and each call takes about a second and a half, almost all of it spent waiting for a response over the network.

Ten thousand documents at three seconds each is over eight hours. During nearly all of that time the process is doing nothing at all: no calculation, no disk work, just waiting for bytes to arrive.

The obvious fix, running many calls at once, has its own failure. The first person to try it wraps everything in threads, fires ten thousand requests, and gets rate limited into a two hour backoff, which is slower than the sequential version and considerably more expensive.

This module covers both halves. Concurrency, so waiting happens in parallel, and control, so you do not overwhelm the service you depend on.

Blocking versus non-blocking, concurrency versus parallelism

Blocking means a call occupies the thread until it returns. time.sleep(2) blocks. requests.get(url) blocks. During those two seconds the thread cannot do anything else, even though it is not using the processor.

Non-blocking means the call yields control while waiting, so something else can run in the meantime, and resumes when the result is available.

Concurrency means several tasks are in progress at once, making progress by interleaving. Parallelism means several tasks execute at literally the same instant on different processor cores.

The practical distinction is what each solves.

Your embedding job is I/O bound: the limit is time spent waiting for the network. Concurrency fixes it, and one core is plenty because the work is waiting rather than computing.

Computing embeddings locally on the processor, or parsing two million documents, is CPU bound: the limit is computation. Concurrency does not help at all, because there is no waiting to overlap. That needs parallelism, which in Python means multiple processes.

asyncio provides concurrency on a single thread. It does not provide parallelism, and using it for CPU-bound work makes things slower rather than faster. Diagnosing which kind of work you have is the first step, and it is usually obvious: if the process is at high processor use, more concurrency will not help.

async, await, coroutines, and the event loop

import asyncio


async def embed_document(doc: Document) -> list[float]:
    response = await client.post(EMBED_URL, json={"input": doc.content})
    return response.json()["embedding"]

async def defines a coroutine function. Calling it does not run the body. It returns a coroutine object, exactly as calling a generator function in Module 4 returned a generator without executing anything.

coro = embed_document(doc)        # nothing has happened yet
result = await coro               # now it runs

await does two things. It runs the awaited operation, and it tells the event loop that this coroutine is pausing and something else may run until the result is ready.

The event loop is the scheduler. It holds a set of tasks, runs one until that task awaits something, then switches to another task that is ready. There is exactly one thread, and only one piece of your code runs at any instant.

That single-threaded property is worth holding onto, because it has a comfortable consequence and an uncomfortable one. The comfortable one is that between two await points your code cannot be interrupted, so many of the data race problems of threaded code do not arise. The uncomfortable one is Lesson 1's central warning: if a task never awaits, nothing else runs at all.

Running a coroutine.

asyncio.run(main())               # from synchronous code, once, at the top level

asyncio.run creates the loop, runs the coroutine, and closes the loop. Call it once at your program's entry point, not inside library code and never inside a running loop.

Awaiting is sequential. This is the mistake everyone makes first:

async def process_all(docs: list[Document]) -> list[list[float]]:
    results = []
    for doc in docs:
        results.append(await embed_document(doc))     # still one at a time
    return results

Every await here waits for completion before the next begins. This is the sequential version with extra syntax, and it is no faster. Concurrency requires launching work before awaiting it.

gather, as_completed, and TaskGroup

asyncio.gather runs coroutines concurrently and returns results in the order given.

embeddings = await asyncio.gather(*(embed_document(doc) for doc in docs))

Note the ordering guarantee: results match the input order regardless of which finished first, which is what makes pairing results with inputs safe.

By default, one exception cancels the remaining tasks and propagates. To collect failures instead:

results = await asyncio.gather(*coros, return_exceptions=True)

succeeded = [r for r in results if not isinstance(r, Exception)]
failed = [r for r in results if isinstance(r, Exception)]

This is the partial results idea from Module 6 applied to concurrency: nine successes and one failure should produce nine results, not zero.

asyncio.as_completed yields results as they finish rather than in order, which is right when you want to start using each result immediately.

for coro in asyncio.as_completed(tasks):
    result = await coro
    await write_result(result)        # begin writing before all are done

The tradeoff is that you no longer know which input a result came from unless the result carries it, which is a good reason to have tasks return a record containing the document id rather than a bare value.

asyncio.TaskGroup is the modern default, available from Python 3.11.

async def process_all(docs: list[Document]) -> list[Embedding]:
    results: list[Embedding] = []
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(embed_document(doc)) for doc in docs]
    return [task.result() for task in tasks]

Its advantage over gather is structured cleanup. The async with block does not exit until every task in the group has finished, and if any task raises, the others are cancelled and the block waits for that cancellation to complete before propagating. You cannot accidentally leave a task running after the function that created it returns, which is a real and confusing bug with bare create_task.

When several tasks fail, TaskGroup raises an ExceptionGroup containing all of them, which is caught with except*:

try:
    async with asyncio.TaskGroup() as group:
        for doc in docs:
            group.create_task(embed_document(doc))
except* ProviderError as eg:
    for exc in eg.exceptions:
        logger.error("provider failure: %s", exc)

Use TaskGroup for work that belongs together and must all complete. Use gather with return_exceptions=True when you want every result including the failures. Use as_completed when order does not matter and early results are useful.

Async context managers and cleanup that survives cancellation

async with is the asynchronous form of with, for resources whose setup or teardown needs to await.

async with httpx.AsyncClient(timeout=timeout) as client:
    response = await client.post(url, json=payload)

The critical property is that cleanup runs even when the block is cancelled. Cancellation in asyncio works by raising asyncio.CancelledError inside the coroutine at its current await point, and because it is an exception, finally blocks and context manager exits run normally.

That gives you the shape for anything that must be released:

async def guarded_generation(semaphore: asyncio.Semaphore):
    await semaphore.acquire()
    try:
        async for chunk in stream_completion():
            yield chunk
    finally:
        semaphore.release()         # runs on success, error, and cancellation

Three rules for cleanup under cancellation.

Do not swallow CancelledError. Catching it and continuing breaks the cancellation and leaves a task running that something is waiting to have stopped. If you catch it to clean up, re-raise.

try:
    await long_operation()
except asyncio.CancelledError:
    await release_resources()
    raise                            # always

Note that CancelledError inherits from BaseException rather than Exception, so except Exception does not catch it. That is deliberate and helpful, and it is another reason Module 5 warned against bare except.

Keep cleanup fast and avoid awaiting in it where possible. A cancelled task performing a slow await during cleanup may itself be cancelled again, and the resource is then genuinely leaked. When cleanup must await something slow, asyncio.shield protects it, though shielding should be rare and deliberate.

Close what you opened. An AsyncClient created without a context manager and never closed leaks connections, and the symptom appears much later as pool exhaustion.

The async trap: one blocking call freezes the loop

This is the failure that makes people conclude async does not work.

async def process(doc: Document) -> Result:
    embedding = await embed(doc)                    # yields, fine
    score = expensive_similarity(embedding)         # blocks for 200ms
    time.sleep(0.1)                                 # blocks for 100ms
    return Result(doc.id, score)

Between the await and the return, this coroutine never yields. For 300 milliseconds the event loop cannot run any other task. With a hundred concurrent tasks each doing this, everything is serialised and your concurrency has bought nothing.

The symptom is distinctive and worth recognising: concurrent code that is barely faster than sequential code, with low processor use if the blocking is a sleep or a synchronous network call, and high processor use if it is computation.

What blocks. time.sleep, any synchronous HTTP library such as requests, synchronous database drivers, file reads and writes, heavy computation, and most third-party SDKs unless they explicitly document async support.

The fixes, in order of preference.

Use the async version. asyncio.sleep rather than time.sleep, httpx.AsyncClient rather than requests, asyncpg rather than psycopg2 in the synchronous mode, as Module 7 set up.

Move blocking work off the loop with an executor, covered next.

Add explicit yield points inside a long computation with await asyncio.sleep(0), which lets other tasks run. This is a stopgap for a loop you cannot easily move, not a design.

[IMAGE PROMPT M8-1
Purpose: Show how a single blocking call inside one coroutine stalls every other task on the event loop.
Visual type: Two-panel execution timeline comparison.
Prompt: A clean educational comparison with two stacked panels sharing a horizontal time axis. Each panel shows four horizontal lanes labelled "Task A", "Task B", "Task C", and "Task D", plus a fifth lane at the bottom labelled "Event loop". The upper panel is headed "All awaits yield" and shows each task lane with short filled segments labelled "running" interleaved with long empty segments labelled "awaiting", arranged so that at any moment exactly one task is running and the others are awaiting, with the event loop lane showing continuous small switching marks labelled "switching between ready tasks". The lower panel is headed "One blocking call" and shows Task A with a single very long filled segment labelled "blocking: time.sleep(0.3)", while Tasks B, C, and D show flat empty segments across the same span labelled "cannot run, not awaiting", and the event loop lane shows one long flat segment labelled "blocked, no switching possible". A caption beneath reads "one thread: if a task does not yield, nothing else runs".
Required elements: Four task lanes plus an event loop lane in both panels, interleaved short segments above, one long blocking segment with three stalled lanes below, the shared time axis, the caption.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, consistent lane heights.
Layout: Two panels stacked vertically aligned to a shared time axis, caption centred beneath.
Text labels: "All awaits yield", "One blocking call", "Task A", "Task B", "Task C", "Task D", "Event loop", "running", "awaiting", "switching between ready tasks", "blocking: time.sleep(0.3)", "cannot run, not awaiting", "blocked, no switching possible", "one thread: if a task does not yield, nothing else runs".
Aspect ratio: 16:9
Accessibility: Convey the difference through segment length, fill, and explicit text labels rather than colour alone.
Avoid: Code screenshots, decorative icons, tiny text, logos, watermarks.
Alt text: Two timelines showing tasks interleaving normally when every await yields, and all tasks stalled when one task makes a blocking call that never yields to the event loop.
END IMAGE PROMPT]

run_in_executor and when threads still win

Some blocking work cannot be made async: a synchronous SDK you do not control, a file read, a compression step. Move it to a thread so the loop keeps running.

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=8)


async def load_document(path: Path) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, path.read_text, "utf-8")

The simpler form for most cases:

content = await asyncio.to_thread(path.read_text, encoding="utf-8")

asyncio.to_thread runs the function in the default thread pool and awaits the result, so from the loop's point of view it is an ordinary await and nothing is blocked.

When threads are the right answer. A blocking library with no async version. File I/O, which has no widely used async form on most platforms. Anything releasing the global interpreter lock while it waits, which most I/O does.

When processes are the right answer. CPU-bound work. Threads in Python do not give parallelism for computation because of the global interpreter lock, so heavy computation belongs in ProcessPoolExecutor or a separate worker. [VOLATILE: work on free-threaded Python is changing this picture. Check the current state before making a definitive claim about the interpreter lock.]

Two cautions. A thread pool has a fixed size, so submitting a thousand blocking calls to an eight-worker pool queues nine hundred and ninety two of them, and that queue is invisible unless you measure it. And code running in a thread cannot touch the event loop directly, so scheduling work back onto it requires asyncio.run_coroutine_threadsafe.

Concept check. Your async pipeline makes an API call, then parses the JSON response, then computes a similarity score over a 1536-dimension vector, then writes a line to a file. Which of those four steps threaten the event loop?

Answer

The API call is fine if it uses an async client, and it is the whole problem if it uses a synchronous one.

JSON parsing is computation, but for a typical response it takes well under a millisecond, and blocking the loop for that long is not worth restructuring for. Blocking is a matter of degree, and the practical threshold is somewhere around a few milliseconds.

The similarity computation depends on size. A single 1536-dimension dot product is microseconds. Scoring one vector against a hundred thousand stored vectors is a different matter and belongs in a thread or a library that releases the lock.

The file write blocks, and in a tight loop it will show up. Use asyncio.to_thread, or batch writes so the blocking happens once per hundred records rather than once per record.