Python Asyncio Basics

Asyncio is Python’s standard library for writing concurrent programs using a single thread. Instead of spinning up multiple threads or processes, asyncio runs an event loop that juggles many functions at once by pausing them exactly at the points where they would otherwise sit idle waiting for something slow, like a network response or a timer. This makes asyncio the standard tool for I/O-bound Python programs – web servers, API clients, chat bots, web scrapers, and database clients – where most of the running time is spent waiting rather than computing.

Overview / How it works

Ordinary Python functions run start-to-finish once called; there is no way to pause one in the middle and let another function run. Asyncio introduces a special kind of function called a coroutine, defined with async def. Calling a coroutine function does not run its body immediately – it returns a coroutine object, a paused computation that has not started yet. That coroutine only actually executes when something drives it forward, either await-ing it directly or handing it to the event loop via asyncio.run() or asyncio.create_task().

The await keyword is the pause button. When a coroutine hits await some_other_coroutine(), execution of the current coroutine suspends at that exact line and control returns to the event loop. The event loop then looks for other work it can make progress on – another task that is ready to run, or a completed timer – and runs that instead. When the awaited operation finishes (for example, a timer set with asyncio.sleep() expires), the event loop resumes the original coroutine right where it left off. This is called cooperative multitasking: coroutines voluntarily give up control at await points instead of being forcibly interrupted the way threads are by the OS scheduler.

Because everything runs on one thread, only one piece of Python code is ever executing at any instant – asyncio gives you concurrency (many things in progress) but not parallelism (many things executing simultaneously on different CPU cores). This is exactly why asyncio shines for I/O-bound work, where the CPU is mostly idle anyway, and is a poor fit for CPU-bound work like heavy number crunching, which needs multiprocessing instead.

Syntax

A minimal asyncio program has three ingredients: a coroutine function, an await expression inside it, and an entry point that starts the event loop.

async def my_coroutine():
    result = await something_awaitable()
    return result

asyncio.run(my_coroutine())
Construct Purpose
async def Defines a coroutine function; calling it returns a coroutine object instead of running the body
await Suspends the current coroutine until the awaited coroutine, task, or future completes
asyncio.run(coro) Creates a new event loop, runs the given coroutine to completion, then closes the loop; the normal program entry point
asyncio.create_task(coro) Schedules a coroutine to run concurrently on the event loop and returns a Task handle immediately, without waiting for it
asyncio.gather(*coros) Runs several awaitables concurrently and returns their results as a list, in the same order they were passed in
asyncio.sleep(seconds) A non-blocking delay – it yields control back to the event loop instead of freezing the whole program

Examples

Example 1: sequential awaits

import asyncio
import time

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)

async def main():
    start = time.perf_counter()
    await say_after(1, "hello")
    await say_after(2, "world")
    elapsed = time.perf_counter() - start
    print(f"finished in {elapsed:.0f} seconds")

asyncio.run(main())

Output:

hello
world
finished in 3 seconds

Each await here blocks this coroutine until that specific call finishes before moving to the next line, so the two one-and-two-second delays add up to three seconds total – awaiting one thing after another is still sequential, even though it uses asyncio.

Example 2: running work concurrently with tasks

import asyncio
import time

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)

async def main():
    start = time.perf_counter()
    task1 = asyncio.create_task(say_after(1, "hello"))
    task2 = asyncio.create_task(say_after(2, "world"))
    await task1
    await task2
    elapsed = time.perf_counter() - start
    print(f"finished in {elapsed:.0f} seconds")

asyncio.run(main())

Output:

hello
world
finished in 2 seconds

asyncio.create_task() schedules both coroutines onto the event loop right away, so their one-second and two-second waits overlap. The program now finishes in roughly two seconds – the length of the longest wait – instead of the sum of both.

Example 3: gathering results from multiple coroutines

import asyncio

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

async def main():
    results = await asyncio.gather(
        fetch_data("users", 2),
        fetch_data("orders", 1),
        fetch_data("products", 3),
    )
    for result in results:
        print(result)

asyncio.run(main())

Output:

Starting fetch: users
Starting fetch: orders
Starting fetch: products
Finished fetch: orders
Finished fetch: users
Finished fetch: products
{'name': 'users', 'delay': 2}
{'name': 'orders', 'delay': 1}
{'name': 'products', 'delay': 3}

All three fetches start immediately, so the three Starting fetch lines print right away in the order they were passed to gather(). They finish in order of their delay – orders (1s), then users (2s), then products (3s) – but gather() still returns results in the original input order, not completion order, which is why the final printed dictionaries are users, orders, products.

Under the hood: what asyncio.run() actually does

  1. asyncio.run(main()) creates a brand-new event loop for this thread.
  2. It wraps main()‘s coroutine object in a Task and schedules it to run.
  3. The loop repeatedly picks a ready task, runs it until it hits an await on something not yet finished, then parks it and moves to the next ready task.
  4. Timers like asyncio.sleep() and I/O readiness (sockets, pipes) are tracked by the loop itself; when a timer expires or a socket becomes readable, the loop marks the waiting task as ready again.
  5. Once the top-level main() task completes, asyncio.run() cancels any leftover background tasks, closes the loop, and returns main()‘s result.

Nothing here uses multiple OS threads – the illusion of concurrency comes entirely from tasks politely handing control back to the loop at every await.

Common Mistakes

Mistake 1: forgetting to await a coroutine

Writing greet() instead of await greet() inside another coroutine does not run greet‘s body at all – it just creates a coroutine object that is immediately discarded, and Python emits a RuntimeWarning: coroutine 'greet' was never awaited. Any code that expected greet‘s side effects (its prints, its return value) silently never happens. Always await a coroutine, pass it to asyncio.create_task(), or pass it to asyncio.gather() – never call it and ignore the result.

Mistake 2: using a blocking call inside a coroutine

Mixing in a synchronous, blocking function like time.sleep() instead of await asyncio.sleep() freezes the entire event loop, not just the current coroutine, because nothing yields control back. This silently destroys any concurrency you were relying on:

import asyncio
import time

async def blocking_task(name, delay):
    print(f"Starting {name}")
    time.sleep(delay)
    print(f"Finished {name}")

async def main():
    start = time.perf_counter()
    await asyncio.gather(
        blocking_task("A", 1),
        blocking_task("B", 1),
    )
    print(f"Total: {time.perf_counter() - start:.0f}s")

asyncio.run(main())

Output:

Starting A
Finished A
Starting B
Finished B
Total: 2s

Even though both tasks were gathered, time.sleep() never yields to the loop, so task B cannot even start until task A’s blocking sleep finishes – the total time is the sum of both delays, not the max. Swapping in asyncio.sleep() fixes it:

import asyncio
import time

async def non_blocking_task(name, delay):
    print(f"Starting {name}")
    await asyncio.sleep(delay)
    print(f"Finished {name}")

async def main():
    start = time.perf_counter()
    await asyncio.gather(
        non_blocking_task("A", 1),
        non_blocking_task("B", 1),
    )
    print(f"Total: {time.perf_counter() - start:.0f}s")

asyncio.run(main())

Output:

Starting A
Starting B
Finished A
Finished B
Total: 1s

Now both one-second waits overlap and the whole thing finishes in about one second, exactly as the earlier examples demonstrated.

Best Practices

  • Use asyncio.run() as your single top-level entry point rather than manually creating and closing event loops.
  • Never call blocking functions (synchronous file I/O, time.sleep(), CPU-heavy loops, blocking HTTP libraries like plain requests) directly inside a coroutine – use async-native libraries, or offload blocking work with asyncio.to_thread().
  • Keep a reference to tasks created with asyncio.create_task(); if a task object is garbage-collected before it finishes, it can be silently cancelled.
  • Prefer asyncio.gather() (or, on Python 3.11+, asyncio.TaskGroup) over manually awaiting a list of tasks one by one – it both runs them concurrently and collects results/errors cleanly.
  • Wrap network calls in asyncio.wait_for() or a timeout context so a single slow dependency can’t hang your whole program forever.
  • Always await or otherwise handle every coroutine you create – an un-awaited coroutine is almost always a bug.
  • Don’t mix asyncio event loops with raw threads unless you specifically need to bridge sync and async code; reach for asyncio.to_thread() or run_in_executor() instead of ad hoc threading.

Practice Exercises

  • Write a coroutine countdown(name, n) that prints name and a number counting down from n to 1, sleeping one second between numbers with asyncio.sleep(1). Run two countdowns concurrently with asyncio.gather() and confirm their output interleaves rather than running one after the other.
  • Write three coroutines that each simulate downloading a file by sleeping for a different number of seconds and then returning a filename string. Use asyncio.gather() to run all three concurrently and print the total elapsed time; verify it is close to the slowest single download, not the sum of all three.
  • Take the blocking mistake example from this lesson and deliberately break it by replacing asyncio.sleep() with time.sleep() in only one of two gathered coroutines. Predict the total run time before running it, then check whether your prediction was correct.

Summary

  • Asyncio provides concurrency on a single thread via an event loop, coroutines, and the async/await keywords.
  • Calling an async def function returns a coroutine object; nothing runs until it is awaited or scheduled.
  • await suspends the current coroutine at that point and lets the event loop run other ready work.
  • Sequential await calls run one after another; asyncio.create_task() and asyncio.gather() let independent coroutines overlap in time.
  • Asyncio helps with I/O-bound waiting, not CPU-bound computation, because only one thread of Python code ever executes at a time.
  • Blocking calls inside a coroutine freeze the entire event loop and silently destroy concurrency – always use async-native equivalents.