async and await

What you'll learn

• Write a coroutine with async def and run it with asyncio.run
• Understand what await does: hand control back to the event loop until a result is ready
• Run several coroutines concurrently with asyncio.gather
• Tell what asyncio is for, and why it needs async libraries all the way down

Concept

Threads overlap waiting by letting the operating system switch between them. asyncio overlaps waiting in one thread by having each task say, explicitly, "I am waiting now, run something else". That statement is await. A coroutine is a function defined with async def; calling it creates a coroutine object that does nothing until an event loop runs it. Inside, await something pauses the coroutine until the awaited thing completes, and the event loop runs other coroutines in the meantime. One thread, many tasks, no locks for the common case, and the switches happen only where you wrote await.

The second half of the story is the loop's side. While both timers run, the loop has nothing to do, so it waits on both at once. When the first timer fires it resumes fetch_a, which finishes, and then fetch_b. Nothing ran in parallel, yet two one-second waits cost one second between them, because waiting is the one thing a single thread can do for many coroutines at the same time.

asyncio.run(main()) starts the loop, runs the coroutine to completion, and shuts the loop down; it is the entry point and you call it once. Inside, await coro() runs one coroutine and waits for it, which is sequential. await asyncio.gather(a(), b(), c()) runs several concurrently and returns their results in order, which is where the speedup comes from. asyncio.sleep(s) is the awaitable stand-in for any wait; time.sleep(s) would block the whole loop, because it does not await.

That last row is the rule that makes or breaks asyncio programs. The event loop only regains control at an await. A blocking call, time.sleep, a synchronous HTTP request, a big CPU loop, freezes every task until it returns. So asyncio needs async libraries all the way down: httpx.AsyncClient instead of httpx.get, async database drivers, asyncio.sleep. When you must call blocking code, await asyncio.to_thread(func, ...) runs it in a thread and awaits the result, which keeps the loop free.

asyncio is the right tool when a program spends its time waiting on many things at once: web servers, API clients that fan out to dozens of endpoints, chat services, scrapers. It is the wrong tool for CPU-bound work, which still needs processes, and it adds ceremony to programs that make one request and exit. Frameworks like FastAPI are built on it, which is why understanding await is part of being production-ready even if you never write a raw event loop.

Code

First coroutines: defining, running, and the difference between awaiting in sequence and gathering. Three one-second waits take one second with gather.

# first_async.py
import asyncio
import time


async def fetch(name, seconds):
    await asyncio.sleep(seconds)
    return f"{name} took {seconds}s"


async def main():
    start = time.perf_counter()
    a = await fetch("a", 0.3)
    b = await fetch("b", 0.3)
    print(a, "|", b, "| sequential ≥ 0.6s:", time.perf_counter() - start >= 0.6)

    start = time.perf_counter()
    results = await asyncio.gather(fetch("x", 0.3), fetch("y", 0.3), fetch("z", 0.3))
    print(results, "| concurrent < 0.5s:", time.perf_counter() - start < 0.5)


coro = main()
print(type(coro).__name__, "created, not yet running")
asyncio.run(coro)

coroutine created, not yet running
a took 0.3s | b took 0.3s | sequential ≥ 0.6s: True
['x took 0.3s', 'y took 0.3s', 'z took 0.3s'] | concurrent < 0.5s: True

Where the switches happen. Each coroutine prints a step and then yields with sleep(0); the interleaving shows the loop moving between them only at await, in a round-robin because both yield every step.

# interleave.py
import asyncio


async def worker(name, steps):
    for i in range(steps):
        print(f"{name} step {i}")
        await asyncio.sleep(0)          # yield to the loop, even with nothing to wait for
    return name


async def main():
    done = await asyncio.gather(worker("fast", 3), worker("slow", 2))
    print("finished:", done)


asyncio.run(main())

fast step 0
slow step 0
fast step 1
slow step 1
fast step 2
finished: ['fast', 'slow']

The blocking mistake and its fix. time.sleep inside a coroutine freezes everything; asyncio.to_thread runs the blocking call elsewhere and keeps the loop responsive.

# blocking.py
import asyncio
import time


async def ticker(label, ticks):
    for _ in range(ticks):
        await asyncio.sleep(0.1)
    return f"{label} ticked"


def blocking_io():
    time.sleep(0.4)
    return "blocking done"


async def wrong():
    start = time.perf_counter()
    results = await asyncio.gather(ticker("t", 3), asyncio.sleep(0, result=blocking_io()))
    return time.perf_counter() - start, results


async def right():
    start = time.perf_counter()
    results = await asyncio.gather(ticker("t", 3), asyncio.to_thread(blocking_io))
    return time.perf_counter() - start, results


elapsed, results = asyncio.run(wrong())
print("blocking in the loop: ticker could not run during sleep; total ≥ 0.7s:", elapsed >= 0.7, results)
elapsed, results = asyncio.run(right())
print("to_thread: overlapped; total < 0.6s:", elapsed < 0.6, results)

blocking in the loop: ticker could not run during sleep; total ≥ 0.7s: True ['t ticked', 'blocking done']
to_thread: overlapped; total < 0.6s: True ['t ticked', 'blocking done']

The tiny JSON server from the HTTP lesson, repeated here so this lesson runs on its own; it is written to disk and not run directly.

# server.py
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

USERS = {1: {"id": 1, "name": "ada", "active": True}, 2: {"id": 2, "name": "linus", "active": False}}


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def send_json(self, status, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        url = urlparse(self.path)
        if url.path == "/users":
            active = parse_qs(url.query).get("active")
            users = list(USERS.values())
            if active:
                users = [u for u in users if u["active"] == (active[0] == "true")]
            return self.send_json(200, users)
        if url.path.startswith("/users/"):
            user = USERS.get(int(url.path.rsplit("/", 1)[1]))
            return self.send_json(200, user) if user else self.send_json(404, {"error": "no such user"})
        if url.path == "/delay":
            time.sleep(0.2)
            return self.send_json(200, {"waited": 0.2})
        if url.path == "/slow":
            time.sleep(2)
            return self.send_json(200, {})
        if url.path == "/boom":
            return self.send_json(503, {"error": "try later"})
        self.send_json(404, {"error": "unknown path"})

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        data = json.loads(self.rfile.read(length) or b"{}")
        if "name" not in data:
            return self.send_json(400, {"error": "name is required"})
        new_id = max(USERS) + 1
        USERS[new_id] = {"id": new_id, "name": data["name"], "active": True}
        self.send_json(201, USERS[new_id])


class Server(ThreadingHTTPServer):
    request_queue_size = 32          # ten connections arriving at once must not overflow the backlog


def start_server():
    server = Server(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return f"http://127.0.0.1:{server.server_port}"

Async all the way down: an async HTTP client against the local server from the HTTP lesson, fanning out ten slow requests concurrently in one thread and timing them against the same ten in a row.

# async_http.py
import asyncio
import time

import httpx

from server import start_server


async def fetch(client, path):
    response = await client.get(path)
    return response.status_code


async def main(base):
    async with httpx.AsyncClient(base_url=base, timeout=5) as client:
        start = time.perf_counter()
        codes = await asyncio.gather(*(fetch(client, "/delay") for _ in range(10)))
        concurrent = time.perf_counter() - start

        start = time.perf_counter()
        for _ in range(10):
            await fetch(client, "/delay")
        sequential = time.perf_counter() - start

        print(codes)
        print("ten together beat ten in a row:", concurrent < sequential)


asyncio.run(main(start_server()))

[200, 200, 200, 200, 200, 200, 200, 200, 200, 200]
ten together beat ten in a row: True

Try it yourself

Write async_pipeline.py with three coroutines that each sleep a different fraction of a second and return a value, and a main that runs them sequentially and then with gather, printing both durations. Add a deliberately blocking time.sleep call to one of them, watch the total jump, then fix it with asyncio.to_thread. Finally, make five concurrent requests to the local HTTP server with httpx.AsyncClient and print the status codes.

Common mistakes

• Forgetting await. fetch() alone creates a coroutine that never runs; Python warns "coroutine was never awaited".
• Blocking inside the loop. time.sleep, requests.get, or heavy computation stalls every task; use the async equivalent or to_thread.
• Calling asyncio.run more than once or inside a running loop. It is the single entry point; everything else is await.

What's next

• async def defines a coroutine; await pauses it and lets the loop run others; asyncio.run is the entry point.
• gather runs coroutines concurrently in one thread; the speedup comes from overlapping waits.
• asyncio needs async libraries throughout; wrap unavoidable blocking calls in to_thread.

You understand the model. Next you will use asyncio the way production code does: task groups, timeouts, async generators and context managers, and clean cancellation.