🟣 Advanced

Async / Await (asyncio)

📖 Lesson 29 ⏱ 45 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

Understand the event loop and cooperative multitasking model
Write coroutines with async def and await
Run coroutines concurrently with asyncio.gather and asyncio.create_task
Use async context managers and async iterators
Handle timeouts and cancellation
Integrate synchronous blocking code with run_in_executor
Understand when asyncio beats threads and when it doesn't

The Problem asyncio Solves

Threads handle concurrency by having the OS switch between them. This works, but threads are expensive (each costs ~8 MB of stack) and synchronisation is error-prone. asyncio takes a different approach: a single thread runs many tasks by cooperatively yielding control whenever it would otherwise block on I/O.

Threading (preemptive — OS switches):
Thread 1: ██▒▒██▒▒██   (▒ = OS-forced context switch overhead)
Thread 2: ▒▒██▒▒██▒▒

asyncio (cooperative — task yields at await):
Task 1:   ██░░██░░██   (░ = waiting for I/O, yields to event loop)
Task 2:   ░░██░░██░░
All on ONE thread — no locking needed!

concurrency_models.txt

asyncio is ideal for I/O-bound workloads with many concurrent tasks — thousands of simultaneous HTTP connections, WebSocket clients, or database queries. A single asyncio event loop can handle far more concurrent connections than a thread pool, with much lower memory overhead. It does not help with CPU-bound work.

Coroutines: `async def` and `await`

A coroutine is a function defined with async def. Inside it, await suspends execution until an awaitable completes — yielding control back to the event loop so other tasks can run.

import asyncio

async def greet(name, delay):
    print(f"Hello, {name}!")
    await asyncio.sleep(delay)   # yields control to the event loop
    print(f"Goodbye, {name}!")

# A coroutine function returns a coroutine OBJECT when called
coro = greet("Alice", 1)
print(type(coro))   # <class 'coroutine'>

# To actually RUN it, schedule it on an event loop
asyncio.run(greet("Alice", 1))
# Hello, Alice!
# (1 second pause)
# Goodbye, Alice!

coroutine_basics.py

asyncio.run() is the entry point — it creates a new event loop, runs the coroutine until it completes, and tears down the loop. Call it once at the top level. Never call it from inside a running event loop (use await or asyncio.create_task() instead).

Sequential vs Concurrent Execution

import asyncio
import time

async def fetch(url, delay):
    print(f"Fetching {url}…")
    await asyncio.sleep(delay)   # simulate network I/O
    print(f"Done: {url}")
    return f"data from {url}"

# ── Sequential: awaiting one at a time ── (~6 seconds)
async def sequential():
    start = time.perf_counter()
    r1 = await fetch("url1", 2)
    r2 = await fetch("url2", 3)
    r3 = await fetch("url3", 1)
    print(f"Sequential: {time.perf_counter()-start:.1f}s")

# ── Concurrent: all start at once ── (~3 seconds — longest single task)
async def concurrent():
    start = time.perf_counter()
    r1, r2, r3 = await asyncio.gather(
        fetch("url1", 2),
        fetch("url2", 3),
        fetch("url3", 1),
    )
    print(f"Concurrent: {time.perf_counter()-start:.1f}s")

asyncio.run(sequential())   # ~6s
asyncio.run(concurrent())   # ~3s

sequential_vs_concurrent.py

Tasks — `asyncio.create_task`

asyncio.gather() is the simplest way to run coroutines concurrently. For more control — cancellation, timeouts, dynamic scheduling — use Tasks:

import asyncio

async def worker(name, delay):
    print(f"{name} starting")
    await asyncio.sleep(delay)
    print(f"{name} done")
    return f"result from {name}"

async def main():
    # create_task schedules coroutines to run concurrently
    # (they start immediately, not just when awaited)
    task1 = asyncio.create_task(worker("A", 2), name="task-A")
    task2 = asyncio.create_task(worker("B", 1), name="task-B")
    task3 = asyncio.create_task(worker("C", 3), name="task-C")

    print("Tasks created — doing other work while they run…")
    await asyncio.sleep(0)   # yield to let tasks start

    # Wait for all and collect results
    results = await asyncio.gather(task1, task2, task3)
    print(results)
    # B finishes first (delay=1), then A (2), then C (3)

asyncio.run(main())

tasks.py

asyncio.gather vs asyncio.wait

import asyncio

async def job(n, delay):
    await asyncio.sleep(delay)
    if n == 2:
        raise ValueError("job 2 failed!")
    return n * 10

async def main():
    # gather — raises on first exception by default
    try:
        results = await asyncio.gather(job(1, 1), job(2, 0.5), job(3, 1.5))
    except ValueError as e:
        print(f"gather raised: {e}")

    # gather with return_exceptions=True — all results, errors as values
    results = await asyncio.gather(
        job(1, 1), job(2, 0.5), job(3, 1.5),
        return_exceptions=True
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"Result: {r}")

asyncio.run(main())

gather_errors.py

Timeouts & Cancellation

import asyncio

async def slow_operation():
    print("Starting slow operation…")
    await asyncio.sleep(10)
    return "done"

async def main():
    # ── asyncio.wait_for: cancel if it takes too long ──
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Timed out!")

    # ── asyncio.timeout (Python 3.11+) ──
    try:
        async with asyncio.timeout(2.0):
            result = await slow_operation()
    except asyncio.TimeoutError:
        print("Timed out again!")

asyncio.run(main())

timeout.py

import asyncio

async def cancellable():
    try:
        print("Working…")
        await asyncio.sleep(5)
        print("Finished")
    except asyncio.CancelledError:
        print("I was cancelled — cleaning up")
        raise   # always re-raise CancelledError!

async def main():
    task = asyncio.create_task(cancellable())
    await asyncio.sleep(1)
    task.cancel()          # request cancellation
    try:
        await task         # wait for it to actually stop
    except asyncio.CancelledError:
        print("Task confirmed cancelled")

asyncio.run(main())
# Working…
# I was cancelled — cleaning up
# Task confirmed cancelled

cancellation.py

Always re-raise CancelledError after cleanup. Swallowing it prevents asyncio from knowing the task was cancelled, breaking the cancellation chain. This is the async equivalent of not swallowing KeyboardInterrupt.

Async Context Managers

Use async with for context managers whose setup or teardown involves I/O (e.g. connecting to a database, acquiring an async lock):

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection(url):
    print(f"Connecting to {url}…")
    await asyncio.sleep(0.1)   # simulate async connect
    conn = {"url": url, "open": True}
    try:
        yield conn
    finally:
        print("Closing connection")
        conn["open"] = False

async def main():
    async with managed_connection("postgres://localhost/db") as conn:
        print(f"Using connection: {conn}")
        await asyncio.sleep(0.1)  # simulate query

asyncio.run(main())

# asyncio also ships with AsyncLock, AsyncSemaphore, etc.
async def safe_write(lock, data):
    async with lock:    # asyncio.Lock — async version
        print(f"Writing: {data}")
        await asyncio.sleep(0.1)

async_cm.py

Async Iterators & Generators

import asyncio

# ── async generator ──
async def ticker(delay, count):
    """Yield numbers with an async delay between each."""
    for i in range(count):
        await asyncio.sleep(delay)
        yield i

# ── async for ── consumes an async iterable
async def main():
    async for tick in ticker(0.2, 5):
        print(f"Tick: {tick}")

    # ── async comprehension ──
    values = [tick async for tick in ticker(0.1, 5)]
    print(values)   # [0, 1, 2, 3, 4]

asyncio.run(main())

async_iter.py

asyncio Synchronisation Primitives

import asyncio

# ── asyncio.Lock ──
lock = asyncio.Lock()

async def protected():
    async with lock:
        print("In protected section")
        await asyncio.sleep(0.1)

# ── asyncio.Event ──
ready = asyncio.Event()

async def setter():
    await asyncio.sleep(1)
    ready.set()

async def waiter():
    await ready.wait()
    print("Event fired!")

# ── asyncio.Semaphore — limit concurrency ──
sem = asyncio.Semaphore(3)

async def limited(n):
    async with sem:
        print(f"Task {n} running")
        await asyncio.sleep(0.5)

async def main():
    # At most 3 tasks run at once
    await asyncio.gather(*[limited(i) for i in range(10)])

asyncio.run(main())

async_sync.py

asyncio.Queue

import asyncio
import random

async def producer(q, n):
    for i in range(n):
        await asyncio.sleep(random.uniform(0.05, 0.2))
        await q.put(i)
        print(f"Produced: {i}")
    await q.put(None)   # sentinel

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(0.1)
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=5)
    await asyncio.gather(producer(q, 10), consumer(q))

asyncio.run(main())

async_queue.py

Running Blocking Code with `run_in_executor`

Blocking (synchronous) code stalls the entire event loop. Use loop.run_in_executor() to offload it to a thread or process pool:

import asyncio
import time

def blocking_io(path):
    """Synchronous file read — blocks the caller."""
    time.sleep(0.5)   # simulate slow disk
    return f"data from {path}"

def cpu_heavy(n):
    """Synchronous CPU work."""
    return sum(i*i for i in range(n))

async def main():
    loop = asyncio.get_event_loop()

    # Run blocking I/O in a thread pool (default executor)
    result = await loop.run_in_executor(None, blocking_io, "file.txt")
    print(result)

    # Convenience wrapper (Python 3.9+)
    result = await asyncio.to_thread(blocking_io, "file2.txt")
    print(result)

    # Run CPU work in a process pool
    from concurrent.futures import ProcessPoolExecutor
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy, 1_000_000)
    print(result)

asyncio.run(main())

run_in_executor.py

asyncio.to_thread() (Python 3.9+) is the clean, modern way to run a blocking function in a thread without touching the event loop directly. Use it whenever you need to call synchronous libraries (like requests, file I/O, or database drivers) from async code.

Common Patterns

Bounded concurrency with Semaphore

import asyncio

async def fetch(session_sem, url):
    async with session_sem:   # limit to N concurrent requests
        await asyncio.sleep(0.2)   # simulate HTTP
        return f"response from {url}"

async def fetch_all(urls, concurrency=10):
    sem     = asyncio.Semaphore(concurrency)
    tasks   = [fetch(sem, url) for url in urls]
    return await asyncio.gather(*tasks)

async def main():
    urls    = [f"https://api.example.com/{i}" for i in range(50)]
    results = await fetch_all(urls, concurrency=5)
    print(f"Fetched {len(results)} URLs")

asyncio.run(main())

bounded_concurrency.py

First result wins — asyncio.wait with FIRST_COMPLETED

import asyncio, random

async def race(name, delay):
    await asyncio.sleep(delay)
    return f"{name} finished!"

async def main():
    tasks = [
        asyncio.create_task(race("A", random.uniform(0.5, 2))),
        asyncio.create_task(race("B", random.uniform(0.5, 2))),
        asyncio.create_task(race("C", random.uniform(0.5, 2))),
    ]

    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    winner = done.pop()
    print(f"Winner: {winner.result()}")

    # Cancel the remaining tasks
    for task in pending:
        task.cancel()

asyncio.run(main())

first_completed.py

asyncio vs Threads vs Processes

Workload	Best tool	Why
Many concurrent I/O tasks (HTTP, DB, WebSocket)	`asyncio`	Single thread, minimal overhead, scales to 10k+ connections
Handful of I/O tasks, existing sync code	`ThreadPoolExecutor`	Simpler to integrate with sync libraries
CPU-bound parallel work	`ProcessPoolExecutor`	True parallelism, bypasses GIL
Blocking library inside async code	`asyncio.to_thread()`	Offloads blocking call without stalling event loop
Real-time streaming / WebSockets	`asyncio`	Native async I/O, no thread-per-connection overhead

Popular async libraries: aiohttp (HTTP client/server), httpx (HTTP client with async mode), asyncpg (PostgreSQL), motor (MongoDB), aiofiles (file I/O), fastapi (web framework — Lesson 37). All of these are designed to work natively with asyncio.

Primary sources: Python Docs — asyncio · Python Docs — Coroutines and Tasks · PEP 492 — Coroutines with async and await

🤖

Ask your AI tutor! Getting RuntimeError: no running event loop? Not sure how to mix sync and async code? Want to understand how await actually suspends execution? asyncio concepts click best with a step-by-step walkthrough.

💻 Exercises

01 Concurrent URL Fetcher

Write an async function fetch_all(urls, concurrency=5) that:

Fetches all URLs concurrently, limiting to concurrency at a time using asyncio.Semaphore
Simulates each request with asyncio.sleep(random.uniform(0.1, 0.5))
Returns a dict mapping each URL to its response (or error string)
Prints progress as each request completes

Compare total time to a sequential version for 20 URLs.

Show solution

import asyncio
import random
import time

async def fetch(sem, url):
    async with sem:
        delay = random.uniform(0.1, 0.5)
        await asyncio.sleep(delay)
        if "error" in url:
            raise ValueError(f"HTTP 500 from {url}")
        result = f"200 OK ({delay:.2f}s)"
        print(f"  ✓ {url}")
        return result

async def fetch_all(urls, concurrency=5):
    sem     = asyncio.Semaphore(concurrency)
    tasks   = {url: asyncio.create_task(fetch(sem, url)) for url in urls}
    results = {}
    for url, task in tasks.items():
        try:
            results[url] = await task
        except Exception as e:
            results[url] = f"ERROR: {e}"
            print(f"  ✗ {url}: {e}")
    return results

async def sequential_fetch(urls):
    results = {}
    for url in urls:
        delay = random.uniform(0.1, 0.5)
        await asyncio.sleep(delay)
        results[url] = f"200 OK ({delay:.2f}s)"
    return results

async def main():
    urls = [f"https://api.example.com/item/{i}" for i in range(20)]
    urls[5] = "https://api.example.com/error/5"

    random.seed(42)
    t = time.perf_counter()
    await sequential_fetch(urls)
    print(f"Sequential: {time.perf_counter()-t:.2f}s")

    random.seed(42)
    t = time.perf_counter()
    results = await fetch_all(urls, concurrency=5)
    print(f"Concurrent: {time.perf_counter()-t:.2f}s")
    print(f"Got {len(results)} results")

asyncio.run(main())

02 Async Rate Limiter

Build an AsyncRateLimiter class that:

Accepts rate (calls per second) in __init__
Has an async def acquire() method that waits if calls would exceed the rate
Can be used as async with limiter: (async context manager)

Test it by firing 10 requests through the limiter at 3 req/s and verifying the timing.

Show solution

import asyncio
import time

class AsyncRateLimiter:
    """Token bucket rate limiter."""

    def __init__(self, rate):
        self.rate      = rate          # calls per second
        self.interval  = 1.0 / rate    # minimum seconds between calls
        self._last     = 0.0
        self._lock     = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now   = asyncio.get_event_loop().time()
            wait  = self._last + self.interval - now
            if wait > 0:
                await asyncio.sleep(wait)
            self._last = asyncio.get_event_loop().time()

    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, *args):
        pass

async def make_request(limiter, n):
    async with limiter:
        t = time.perf_counter()
        await asyncio.sleep(0.01)   # simulate fast API call
        print(f"Request {n:2d} at t={t:.3f}s")
        return n

async def main():
    limiter = AsyncRateLimiter(rate=3)   # 3 requests per second
    start = time.perf_counter()
    results = await asyncio.gather(*[make_request(limiter, i) for i in range(10)])
    elapsed = time.perf_counter() - start
    print(f"\n{len(results)} requests in {elapsed:.2f}s "
          f"(expected ≥ {(len(results)-1)/3:.2f}s at 3 req/s)")

asyncio.run(main())

03 Async Pipeline

Build a three-stage async pipeline using asyncio.Queue:

Stage 1 — Generator: produces numbers 1–15 with a 0.05s delay each
Stage 2 — Transformer (2 workers): takes a number, computes n² with a 0.1s delay
Stage 3 — Collector: receives results, accumulates them, prints a summary

Use sentinel values to signal shutdown at each stage. Verify total time is less than sequential.

Show solution

import asyncio
import time

NUM_WORKERS = 2
SENTINEL    = None

async def generator(out_q, items):
    for item in items:
        await asyncio.sleep(0.05)
        await out_q.put(item)
        print(f"  Gen  → {item}")
    # One sentinel per worker
    for _ in range(NUM_WORKERS):
        await out_q.put(SENTINEL)

async def transformer(worker_id, in_q, out_q):
    while True:
        item = await in_q.get()
        if item is SENTINEL:
            await out_q.put(SENTINEL)
            break
        await asyncio.sleep(0.1)   # simulate work
        result = item ** 2
        print(f"  W{worker_id}   → {item}² = {result}")
        await out_q.put(result)
        in_q.task_done()

async def collector(in_q, num_workers):
    results = []
    finished = 0
    while finished < num_workers:
        item = await in_q.get()
        if item is SENTINEL:
            finished += 1
        else:
            results.append(item)
    print(f"\nCollected {len(results)} results")
    print(f"Sum of squares: {sum(results)}")
    return results

async def main():
    work_q    = asyncio.Queue(maxsize=5)
    results_q = asyncio.Queue()

    start = time.perf_counter()

    await asyncio.gather(
        generator(work_q, range(1, 16)),
        *[transformer(i, work_q, results_q) for i in range(NUM_WORKERS)],
        collector(results_q, NUM_WORKERS),
    )

    print(f"\nPipeline completed in {time.perf_counter()-start:.2f}s")
    # Sequential would take 15*0.05 + 15*0.1 = 2.25s
    # Pipeline overlaps stages: ~(15*0.05) + (15/2 * 0.1) ≈ 1.5s

asyncio.run(main())

Async / Await (asyncio)

🎯 Learning Objectives

The Problem asyncio Solves

Coroutines: async def and await

Sequential vs Concurrent Execution

Tasks — asyncio.create_task

asyncio.gather vs asyncio.wait

Timeouts & Cancellation

Async Context Managers

Async Iterators & Generators

asyncio Synchronisation Primitives

asyncio.Queue

Running Blocking Code with run_in_executor

Common Patterns

Bounded concurrency with Semaphore

First result wins — asyncio.wait with FIRST_COMPLETED

asyncio vs Threads vs Processes

💻 Exercises

Coroutines: `async def` and `await`

Tasks — `asyncio.create_task`

Running Blocking Code with `run_in_executor`