Async Python — asyncio, async/await

A Third Way to Handle I/O-Bound Work

Module 34 ended with a decision framework and a deliberate loose thread: for I/O-bound work, it recommended "threading, or asyncio," without fully explaining the second option. This module closes that gap. Asynchronous programming solves the exact same category of problem threading solves — making productive use of time your program would otherwise spend idly waiting on I/O — but with a fundamentally different mechanism, and, for large numbers of concurrent operations, real advantages over threading.

Recall from Module 34 that each OS thread carries real overhead — memory for its own stack, and cost for the operating system to schedule and switch between threads. Fine for a handful of concurrent threads; expensive if you genuinely need thousands of concurrent I/O operations, as a high-traffic web server or a service polling hundreds of external APIs might. Asynchronous programming achieves concurrency within a single thread, using far lighter-weight units called coroutines instead of OS threads — which is exactly why it scales to far larger numbers of concurrent operations than threading comfortably can.

Threading:  concurrency via multiple OS threads, each with real overhead.
Asyncio:    concurrency via many lightweight coroutines, cooperating on ONE thread.

Both exist to solve the same problem — productive waiting during I/O —
using genuinely different mechanisms with different trade-offs.

async def and await — Functions That Can Pause

A coroutine is a special kind of function, defined with async definstead of plain def, that can be paused at specific points and resumed later — without blocking the rest of the program while it waits. Calling a coroutine function does not run its body immediately the way calling a normal function does; it returns a coroutine object, which needs to actually be run by something else (covered in Part 03).

import asyncio

async def greet(name):
    print(f"Hello, {name}")
    await asyncio.sleep(1)     # pauses THIS coroutine for 1 second, without blocking others
    print(f"Goodbye, {name}")

result = greet("Maria")
print(type(result))
# <class 'coroutine'> — calling greet("Maria") did NOT run the function body yet.
# It just created a coroutine object describing the work to be done.

The await keyword is where a coroutine can genuinely pause — it can only appear inside a function defined with async def, and it marks a point where the current coroutine says, in effect, "I am waiting on something; let something else run while I wait." asyncio.sleep() above is the async equivalent of time.sleep(), but crucially it does not block the entire program the way time.sleep() does — it only pauses the one coroutine that called it, freeing everything else to keep running during that wait.

What Actually Schedules and Runs All These Coroutines

Coroutine objects do nothing by themselves — something has to actually run them, deciding which coroutine gets to execute at any given moment and switching between them at their await points. That something is the event loop, the core engine behind asyncio. Understanding it properly, not just treating it as magic, matters for reasoning correctly about async code.

The event loop implements a model called cooperative multitasking. This is a genuinely important contrast with threading, which uses preemptive multitasking — the operating system can interrupt a thread at essentially any point and switch to another one, whether that thread is ready to be interrupted or not. In cooperative multitasking, a coroutine keeps running uninterrupted until it voluntarily yields control, which only happens at an await point. Nothing else can force a coroutine to pause mid-execution — it decides for itself, at each await, whether to hand control back to the event loop.

1. Maintain a set of coroutines that are ready to run, or currently waiting on
   something (a timer, a network response, a file read).
2. Run a ready coroutine until it hits an "await" and voluntarily pauses.
3. While that coroutine is waiting (e.g. waiting on network I/O), run
   whichever OTHER coroutine is ready, if any.
4. When the thing being awaited completes, mark that coroutine as ready again.
5. Repeat, continuously, until there is no more work to do.

This explains precisely why the "one thread avoids blocking" concurrency in Part 01 works: there genuinely is only one thread doing the actual executing, at any given instant — but because coroutines voluntarily step aside at each await, the single thread never sits fully idle waiting on one operation while other work could be progressing. It is a very similar underlying idea to the GIL-released-during-I/O behavior from Module 34, but achieved deliberately and explicitly through await points, rather than as a side effect of the interpreter's memory-management locking.

import asyncio

async def main():
    print("Starting")
    await asyncio.sleep(1)
    print("Done")

asyncio.run(main())
# asyncio.run() creates a fresh event loop, runs the given coroutine to
# completion, and cleans up the event loop afterward — this is the standard
# entry point for a Python script that uses asyncio.

asyncio.gather() — The Whole Point of Doing Any of This

A single await on its own does not give you concurrency — it just pauses the current coroutine until the awaited thing finishes, similar in spirit to a regular blocking call, just without blocking other coroutines that might be running elsewhere. Real concurrency comes from starting several coroutines and letting the event loop interleave their waiting periods. asyncio.gather() is the standard way to do exactly that.

import asyncio
import time

async def fetch_data(name, seconds):
    print(f"Starting {name}")
    await asyncio.sleep(seconds)
    print(f"Finished {name}")
    return f"{name} result"

async def main():
    start = time.time()
    a = await fetch_data("A", 2)
    b = await fetch_data("B", 2)
    c = await fetch_data("C", 2)
    print(f"Total: {time.time() - start:.1f}s")   # ~6 seconds — these ran one after another!

asyncio.run(main())

import asyncio
import time

async def fetch_data(name, seconds):
    print(f"Starting {name}")
    await asyncio.sleep(seconds)
    print(f"Finished {name}")
    return f"{name} result"

async def main():
    start = time.time()
    results = await asyncio.gather(
        fetch_data("A", 2),
        fetch_data("B", 2),
        fetch_data("C", 2),
    )
    print(f"Total: {time.time() - start:.1f}s")   # ~2 seconds — all three ran concurrently!
    print(results)   # ['A result', 'B result', 'C result'] — in the order passed to gather

asyncio.run(main())

Notice the crucial difference: sequentially await-ing three coroutines, one after another, gets you correct code but no speed benefit — each await fully completes before the next one even starts. asyncio.gather() starts all three coroutines and lets the event loop interleave their waiting periods, so the total time is roughly the longest individual wait, not the sum of all of them. This is the entire value proposition of async programming in one comparison.

Fetching From Multiple APIs Concurrently

Here is a realistic worked example: a fictional US travel-booking aggregator needs to query three different airline partner APIs for flight prices on the same route, and combine the results for the user. This is precisely the kind of I/O-bound, "wait on several independent network calls" workload async programming is built for.

import time

def get_prices_sync(airline, delay):
    print(f"Querying {airline}...")
    time.sleep(delay)   # standing in for a real network call to the airline's API
    return {"airline": airline, "price": 210 + delay * 10}

def get_all_prices_sync():
    start = time.time()
    results = [
        get_prices_sync("SkyWest Air", 1.5),
        get_prices_sync("Continental Express", 2.0),
        get_prices_sync("Rocky Mountain Airways", 1.2),
    ]
    print(f"Sequential total: {time.time() - start:.1f}s")   # ~4.7 seconds
    return results

get_all_prices_sync()

import asyncio
import time

async def get_prices_async(airline, delay):
    print(f"Querying {airline}...")
    await asyncio.sleep(delay)   # standing in for a real "async" network call
    return {"airline": airline, "price": 210 + delay * 10}

async def get_all_prices_async():
    start = time.time()
    results = await asyncio.gather(
        get_prices_async("SkyWest Air", 1.5),
        get_prices_async("Continental Express", 2.0),
        get_prices_async("Rocky Mountain Airways", 1.2),
    )
    print(f"Concurrent total: {time.time() - start:.1f}s")   # ~2.0 seconds — the SLOWEST call, not the sum
    return sorted(results, key=lambda r: r["price"])

asyncio.run(get_all_prices_async())

On a real route with real airline partner APIs (each typically responding in a few hundred milliseconds to a couple of seconds), the difference between querying three, five, or ten partners sequentially versus concurrently is the difference between a booking search that feels sluggish and one that feels instant — exactly the kind of user-facing improvement that makes async worth adopting for genuinely I/O-heavy features like this one.

A production version would use a real async HTTP client — the popular third-party library httpx supports async requests directly, and Module 37's requests library (synchronous by design) is not usable inside a coroutine without blocking the whole event loop, exactly the mistake covered next.

The Three Mistakes That Catch Almost Everyone New to asyncio

Forgetting await

Calling a coroutine function without await-ing it does not raise an error immediately — it just creates a coroutine object and does nothing with it, which is a genuinely confusing silent failure.

async def save_to_database(record):
    await asyncio.sleep(0.1)
    print(f"Saved {record}")

async def main():
    save_to_database("order-123")   # BUG: missing "await" — this does NOTHING
    print("Done")

asyncio.run(main())
# Output:
# Done
# (nothing was ever saved — save_to_database's body never actually ran)
#
# Python DOES warn about this: "RuntimeWarning: coroutine 'save_to_database'
# was never awaited" — always pay attention to that specific warning.

Mixing blocking and async code

Calling a genuinely blocking, synchronous function from inside a coroutine — without await, because a normal blocking function has no await to give — does not pause just that one coroutine. It blocks the entire event loop, on the single thread everything is running on, stalling every other coroutine that was supposed to be making progress during that wait.

import time
import asyncio

async def bad_fetch():
    print("Starting bad_fetch")
    time.sleep(3)          # BLOCKING call — freezes the entire event loop for 3 seconds
    print("Finished bad_fetch")

async def good_task():
    print("good_task running")

async def main():
    await asyncio.gather(bad_fetch(), good_task())
    # good_task does NOT get a chance to interleave — bad_fetch's time.sleep(3)
    # blocks the single thread outright, so nothing else can run during those 3 seconds,
    # completely defeating the purpose of using asyncio.gather() in the first place

asyncio.run(main())

Not using asyncio.gather() and losing the concurrency benefit entirely

Covered fully in Part 04 — sequential await calls are valid code, they simply provide none of the speed benefit async programming exists to offer. This is less a "bug" than a missed opportunity, but it is common enough, especially when refactoring existing sequential code to be "async," to include here explicitly.

An Honest Answer, Not a Universal Recommendation

Async programming is a genuinely powerful tool, and it is also genuinely more complex to reason about correctly than synchronous code — the event loop, coroutine vs regular functions, the blocking-call trap from Part 06, and libraries that need async-native versions to work correctly, are all real additional cognitive overhead. It is worth being direct about when that overhead is worth paying.

- Handling a LARGE number of concurrent I/O-bound operations (many simultaneous
  network requests, a web server handling thousands of concurrent connections)
- Building on a framework that is already async-native (FastAPI, for example,
  which you may encounter building APIs professionally)
- I/O-bound work where thread overhead specifically becomes the bottleneck

- A script that makes a handful of sequential API calls, run occasionally,
  where a few extra seconds of total runtime genuinely does not matter
- CPU-bound work — async provides NO benefit here, same underlying reason
  threading doesn't (Module 34) — reach for multiprocessing instead
- Small internal tools and one-off scripts, where the added complexity of
  getting async code correct is not worth the modest speed gain

A genuinely useful gut check: if you are reaching for async because a specific, measured performance problem exists — a batch job that is too slow because it makes many sequential API calls, or a server that needs to handle far more concurrent connections than threading comfortably supports — it is very likely worth the complexity. If you are reaching for it because it feels like the "modern" or "advanced" way to write Python, that instinct alone is not a strong enough reason, and plain synchronous code, or the simpler threading approach from Module 34, is very often the better engineering decision.

A Portland Travel Startup's Frozen Search Page

Four Misconceptions About Async Python

5 Interview Questions — With Complete Answers

Async Mistakes Beginners Make Constantly

Errors You Will Hit With asyncio — And Exactly Why