Waiting Without Blocking

Warm-up: from Day 5

The analyzer takes a break today, but first, three from yesterday — from memory.

1. Predict all three lines. The first is Day 5’s silent miscount.

Warm-up: predict Python

from enum import IntEnum

class Level(IntEnum):
    INFO = 20
    ERROR = 40
    CRITICAL = 50

records = [Level.INFO, Level.ERROR, Level.CRITICAL]
print(sum(1 for r in records if r == "ERROR"))
print(sum(1 for r in records if r >= Level.ERROR))
print(Level.ERROR.name, int(Level.ERROR))

Check your prediction

0
2
ERROR 40

An IntEnum is never equal to a string, so the first count is quietly zero — no error, just a wrong number. Comparing Levels by severity counts both ERROR and CRITICAL. From The Silent Miscount.

2. Fix. When there are no errors, this should print no errors. It crashes.

Warm-up: fix Python

def first_error(levels):
    for i, level in enumerate(levels):
        if level == "ERROR":
            return i
    return None

where = first_error(["INFO", "WARN"])
print(f"first error at line {where + 1}")

Check your fix

def first_error(levels):
    for i, level in enumerate(levels):
        if level == "ERROR":
            return i
    return None

where = first_error(["INFO", "WARN"])
if where is None:
    print("no errors")
else:
    print(f"first error at line {where + 1}")

no errors

The honest hint for this function is -> int | None, and that | None is a promise the caller has to handle. A type checker would have flagged where + 1 before the program ever ran. From Type Hints That Help.

3. Explain. No code: why does Level.from_text raise ValueError for unknown text, rather than letting the KeyError from Level[name] escape?

Check your explanation

Unknown text is a bad value from outside, and the caller — parse_record — already turns ValueError into a ParseError with a line number. A KeyError would leak how the lookup happens to be implemented, and slip past that handling. From The Silent Miscount.

A standalone day

Today is a standalone day. The analyzer takes a break, and you’ll work on a problem it’s about to have: its logs live on several servers, and fetching each one means waiting on the network.

Waiting is the key word. Downloading a log takes a server a few milliseconds of work and your program most of a second of doing nothing. Fetch three logs one after another and you wait three times. Nothing about the second fetch needs the first to be finished. You could be waiting for all three at once.

One at a time, then all at once

Here’s a simulated fetch. asyncio.sleep stands in for the network: it waits without doing any work, which is exactly what a download mostly does.

Three fetches, two ways Python

import asyncio
import time

DELAYS = {"auth": 0.3, "payments": 0.2, "search": 0.1}    # seconds each server takes


async def fetch_log(server):
    """Pretend to download a server's log: mostly waiting on the network."""
    await asyncio.sleep(DELAYS[server])
    return [f"2026-09-01 09:00:00 INFO {server}: ok"] * 3


async def one_at_a_time():
    return [await fetch_log(server) for server in DELAYS]


async def all_at_once():
    return await asyncio.gather(*(fetch_log(server) for server in DELAYS))


for strategy in (one_at_a_time, all_at_once):
    start = time.perf_counter()
    logs = asyncio.run(strategy())
    print(f"{strategy.__name__:13} {len(logs)} logs in {time.perf_counter() - start:.1f}s")

One at a time takes the sum of the waits, 0.6 seconds. All at once takes the longest wait, 0.3 seconds. With fifty servers it would be 50 waits against 1.

The pieces

async def defines a coroutine function. Calling it doesn’t run its body. It creates a coroutine, a paused computation that can be run later:

Calling is not running Python

import asyncio


async def fetch_log(server):
    await asyncio.sleep(0.1)
    return [f"{server}: ok"]


coroutine = fetch_log("auth")
print(coroutine)                      # calling it made a coroutine; nothing has run
print(asyncio.run(coroutine))        # running it gives the return value

await runs a coroutine and gives you its result. It also marks the places where this coroutine is willing to pause. While it’s paused, waiting for the sleep or the network, other coroutines get to run.

asyncio.run(...) starts the event loop, runs one coroutine to completion, and stops the loop. It’s the bridge from ordinary code to async code, and a program usually calls it once, at the top.

asyncio.gather(a, b, c) runs several coroutines at the same time, and returns their results in the order you passed them, whichever finished first.

The event loop is a scheduler. It keeps a list of coroutines that are ready to continue, runs one until it reaches an await that has to wait, then switches to the next. When a sleep ends or data arrives, the coroutine that was waiting becomes ready again.

Concurrent, not parallel

Everything runs on one thread. Only one coroutine is ever running at a given instant. What overlaps is the waiting. That makes asyncio a great fit for network calls, and no help at all for heavy calculation, where no one is ever waiting. The next lesson shows what heavy calculation does to an event loop.

Bug hunt: the log that wasn’t a list

How many lines? Python

import asyncio


async def fetch_log(server):
    await asyncio.sleep(0.1)
    return [f"2026-09-01 09:00:00 INFO {server}: ok"] * 3


async def main():
    lines = fetch_log("auth")
    print(f"auth sent {len(lines)} lines")


asyncio.run(main())

Reproduce. TypeError: object of type 'coroutine' has no len(). One call, one line.

Inspect. The error names the type, but check what lines really is. Replace the print with print(lines). It’s <coroutine object fetch_log at 0x…>. Look at the rest of the output, too. Python may add RuntimeWarning: coroutine 'fetch_log' was never awaited.

Before you open this, write the regression test first. Careful: the bug is in main, so the test has to run main, not just fetch_log. Then fix it and compare.

Show the diagnosis and the fix

Narrow. fetch_log("auth") made a coroutine and never ran it. There’s no await, so there’s no result, just the paused computation, and it’s thrown away unrun at the end of main. That’s what the warning is telling you.

Pin it. Test the code that was broken, main, not fetch_log, which was fine all along. A test that awaited fetch_log correctly would pass while the bug sat in its caller. Capture what main prints:

def test_main_reports_how_many_lines_arrived(self):
    output = io.StringIO()
    with redirect_stdout(output):
        asyncio.run(main())
    self.assertEqual(output.getvalue(), "auth sent 3 lines\n")

Run it before fixing anything, and watch it fail with the same TypeError.

Fix. lines = await fetch_log("auth").

The warning is the clue

A missing await doesn’t always fail as loudly as this. If you’d only printed lines, the program would have printed a coroutine object and carried on. Treat coroutine ... was never awaited as an error every time you see it. A type checker catches it too, once fetch_log has a return type: calling an async def gives a coroutine, and mypy knows a coroutine has no len().

Your turn

Exercise

Check every service at once

check_all asks each service whether it’s healthy, one at a time. Make it ask them all at once, so the second test passes, and keep the result a dict in the same order as services.

The second test doesn’t use a stopwatch. ping counts how many checks are in progress at once, and the test looks at the highest count it reached. One at a time, that’s 1. All at once, it’s 3, however fast or slow your computer is.

import asyncio
import sys
import unittest

SERVICES = {"auth": 0.3, "payments": 0.3, "search": 0.3}
in_progress = 0                         # checks started and not yet finished
most_at_once = 0                        # the highest in_progress ever reached


async def ping(service):
    """Pretend to ask a service if it's healthy."""
    global in_progress, most_at_once
    in_progress += 1
    most_at_once = max(most_at_once, in_progress)
    await asyncio.sleep(SERVICES[service])
    in_progress -= 1
    return service != "search"          # search is down today


async def check_all(services):
    """{service: healthy?} for every service."""
    return {service: await ping(service) for service in services}


class TestCheckAll(unittest.TestCase):
    def test_every_service_is_checked(self):
        self.assertEqual(asyncio.run(check_all(SERVICES)),
                         {"auth": True, "payments": True, "search": False})

    def test_the_checks_overlap(self):
        global most_at_once
        most_at_once = 0
        asyncio.run(check_all(SERVICES))
        self.assertEqual(most_at_once, 3, "the checks ran one at a time")


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestCheckAll)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

Hint 1: the idea

An await inside the loop finishes each check before the next one starts. To overlap them, start every check first and then wait for all of them together — that’s what asyncio.gather does. It hands results back in the order you passed the checks in, whatever order they finish in.

Hint 2: a first step

Make the checks without awaiting them: [ping(s) for s in services]. Then await asyncio.gather(*checks) gives you a list of results in the same order as services, ready to be paired back up with the names.

Show one solution

async def check_all(services):
    """{service: healthy?} for every service, all checked at once."""
    results = await asyncio.gather(*(ping(service) for service in services))
    return dict(zip(services, results))

gather returns results in the order of its arguments, so zip pairs each result with the right service, even though they finish in no particular order.

The original was a dict comprehension with await inside. That’s legal, but each await finishes before the next one starts, so it’s sequential. To run things at the same time, you have to start them all before waiting for any, and that’s the job gather does.

The tests call asyncio.run inside an ordinary TestCase. On your machine, unittest.IsolatedAsyncioTestCase lets test methods be async def directly. It can’t start its own event loop in this browser, so this course uses asyncio.run instead.

What you can now do

Explain why overlapping waits makes three fetches take as long as one.
Tell a coroutine function, a coroutine and its result apart.
Use asyncio.run, await and asyncio.gather.
Spot a missing await from its symptoms: a coroutine object, and a “never awaited” warning.

Next: starting tasks, collecting results as they arrive, and what happens when one coroutine refuses to wait.