Python asyncio from Scratch

Learn how Python's asyncio works, from coroutines to real HTTP calls

Samod Alex
•
4 min read
Python asyncio from Scratch

asyncio confuses people for one reason: they learn the syntax before understanding why it exists. Once you see the problem it solves, everything else follows naturally.


The Problem: Your Program Spends Most of Its Time Waiting

Here's a synchronous function that simulates fetching data from an API:

import time

def fetch(id):
    print(f"Fetching {id}...")
    time.sleep(1)  # network call
    print(f"Done {id}")

start = time.perf_counter()
fetch(1)
fetch(2)
fetch(3)
elapsed = time.perf_counter() - start
print(f"Total: {elapsed:.2f}s")
# Total: 3.00s

Three one-second calls take three seconds. The program isn't busy during those three seconds. It's just waiting. asyncio lets you use that waiting time to run other things.


Coroutines: async def and await

A coroutine is a function defined with async def. Calling it doesn't run it; it returns a coroutine object that the event loop knows how to schedule.

import asyncio

async def greet():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

The await keyword suspends greet at that line and hands control back to the event loop. Other coroutines can run during that pause. Once the sleep finishes, greet picks up exactly where it left off.


Running a Coroutine: asyncio.run()

You can't run a coroutine by calling it like a regular function. You need an event loop. asyncio.run() handles that:

import asyncio

async def greet():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(greet())
# Hello
# World

asyncio.run() creates an event loop, runs the coroutine to completion, then closes the loop. It's the standard entry point for any async program. Use it once, at the top level.


Concurrency with asyncio.gather()

This is where asyncio earns its place. gather() runs multiple coroutines at the same time:

import asyncio
import time

async def fetch(id):
    print(f"Start {id}")
    await asyncio.sleep(1)
    print(f"Done {id}")

async def main():
    start = time.perf_counter()
    await asyncio.gather(fetch(1), fetch(2), fetch(3))
    elapsed = time.perf_counter() - start
    print(f"Finished in {elapsed:.2f}s")

asyncio.run(main())
# Start 1
# Start 2
# Start 3
# Done 1
# Done 2
# Done 3
# Finished in 1.00s

All three start immediately. While fetch(1) is sleeping, fetch(2) and fetch(3) are also sleeping. Total time is ~1 second, not 3.


asyncio.gather() Preserves Order

gather() returns results in the same order you passed the coroutines, regardless of which finished first:

import asyncio

async def fetch(id):
    await asyncio.sleep(0.5)
    return f"Result {id}"

async def main():
    results = await asyncio.gather(
        fetch(1),
        fetch(2),
        fetch(3),
    )
    print(results)
    # ['Result 1', 'Result 2', 'Result 3']

asyncio.run(main())

This makes it safe to unpack results in a predictable way without tracking which finished when.


asyncio.create_task() for Background Work

gather() waits for everything before continuing. If you want to fire off a task and keep going without blocking, use create_task():

import asyncio

async def background_job():
    await asyncio.sleep(2)
    print("Background done")

async def main():
    task = asyncio.create_task(background_job())
    print("Main is doing other things")
    await asyncio.sleep(1)
    print("Main is still going")
    await task
    print("All done")

asyncio.run(main())
# Main is doing other things
# Main is still going
# Background done
# All done

create_task() schedules the coroutine immediately but doesn't block. The await task at the end ensures the job finishes before main() exits. Without it, the program exits and the task is cancelled.


Real-World: Async HTTP with httpx

asyncio.sleep() simulates I/O. Real programs hit APIs. httpx is the cleanest async HTTP client available:

pip install httpx
import asyncio
import httpx
import time

async def fetch_user(client, user_id):
    response = await client.get(
        f"https://jsonplaceholder.typicode.com/users/{user_id}"
    )
    data = response.json()
    return data["name"]

async def main():
    start = time.perf_counter()
    async with httpx.AsyncClient() as client:
        names = await asyncio.gather(
            fetch_user(client, 1),
            fetch_user(client, 2),
            fetch_user(client, 3),
        )
    elapsed = time.perf_counter() - start
    print(names)
    # ['Leanne Graham', 'Ervin Howell', 'Clementine Bauch']
    print(f"Fetched in {elapsed:.2f}s")
    # Fetched in 0.43s

asyncio.run(main())

Three real network calls, concurrent, under half a second. The equivalent requests code would take around 1.3 seconds. The async with block ensures the client connection pool is properly closed when done.


Mistakes That Will Bite You

1. Forgetting await

async def main():
    asyncio.sleep(1)        # wrong: returns a coroutine object and does nothing
    await asyncio.sleep(1)  # correct

Calling a coroutine without await doesn't crash immediately. Python emits a RuntimeWarning: coroutine was never awaited and silently skips it. It's a quiet bug that's hard to catch without warnings enabled.

2. Blocking the event loop

import time

async def bad():
    time.sleep(2)           # blocks the entire event loop
    await asyncio.sleep(2)  # correct: yields control while waiting

Any blocking call inside an async function freezes the whole program. Nothing else can run during that time. If you're calling a slow database driver or doing CPU-heavy computation, use asyncio.to_thread() to run it in a thread pool without blocking:

import asyncio
import time

def slow_sync():
    time.sleep(1)
    return "done"

async def main():
    result = await asyncio.to_thread(slow_sync)
    print(result)
    # done

asyncio.run(main())

3. asyncio.run() inside a running event loop

Calling asyncio.run() from inside Jupyter or FastAPI raises RuntimeError: This event loop is already running. In Jupyter, just await main() directly. In FastAPI, define your route as async def and FastAPI handles the loop.


asyncio vs Threading vs Multiprocessing

Situation

Use

I/O-bound: APIs, databases, file reads

asyncio

Legacy blocking code you can't change

threading

CPU-bound: image processing, ML, math

multiprocessing

asyncio is single-threaded. It doesn't run things in parallel; it runs them concurrently. Parallel means two things happening at the exact same instant on separate cores. Concurrent means one thread switching between tasks so efficiently it feels simultaneous.

For I/O-bound work, that distinction doesn't matter. The wait is real time, and asyncio uses it. For CPU-bound work, asyncio buys you nothing because there's no waiting to exploit.


Summary

Concept

What it does

async def

Defines a coroutine

await

Suspends the coroutine, yields to the event loop

asyncio.run()

Entry point; creates the event loop and runs a coroutine

asyncio.gather()

Runs multiple coroutines concurrently, returns results in order

asyncio.create_task()

Schedules a coroutine without blocking the current one

asyncio.sleep()

Async-safe sleep; yields to the event loop

asyncio.to_thread()

Runs blocking sync code in a thread pool safely

asyncio isn't magic. It's a scheduler that uses your program's waiting time to run other things. Once that clicks, the syntax stops looking mysterious.


Building something with asyncio? Drop it in the comments.

Comments (0)

Join the discussion by logging into your account.

No comments yet. Be the first to comment!

Samod Alex

Passionate developer sharing knowledge about modern web technologies and best practices.

Subscribe to Samod Alex's Newsletter

Direct email dispatches when new stories are published. Zero algorithms.