Asyncio Explained With 3 Real Projects

Quickstart

Want the fastest path to “I get it”? Do this in order. Each step is small, but together they teach you the core asyncio habits: await I/O, limit concurrency, handle timeouts/cancellation, and shut down cleanly.

Overview

Python’s asyncio is a single-threaded concurrency model built around an event loop. Instead of spawning a thread per request, you write code that cooperatively yields control whenever it hits an awaitable I/O operation. That lets one process handle thousands of sockets efficiently — as long as you avoid blocking calls.

The goal isn’t to memorize APIs. It’s to build the mental model: the event loop runs tasks, tasks pause at await, and your job is to keep the loop responsive while managing concurrency and failures.

Core concepts

Asyncio clicks when you stop thinking “my function runs from top to bottom” and start thinking “my function is a task that takes turns with other tasks”. Here are the pieces you’ll meet in every real async program.

Coroutines and await

An async function returns a coroutine object. Nothing inside it runs until you await it (directly or indirectly via a task). When you await an awaitable (like an HTTP request), your coroutine yields control back to the event loop so other tasks can run.

Event loop

The event loop is the scheduler. It watches sockets and timers, wakes up tasks whose I/O completed, and runs them until they hit the next await. If you block the loop (CPU loop, blocking I/O, time.sleep), you block everything.

Tasks: “run this coroutine in the background”

A Task wraps a coroutine and schedules it. You create tasks when you want multiple coroutines to progress concurrently. In modern Python, prefer structured concurrency with asyncio.TaskGroup (3.11+) because it makes lifetimes explicit: if a child task fails, the group cancels the others and bubbles up the error.

Timeouts, cancellation, and why they matter

Real networks fail. A robust asyncio program treats timeouts and cancellation as normal control flow: set deadlines, catch errors, and shut down without leaving dangling tasks.

Step-by-step

We’ll build three “real enough” asyncio projects. Each one is small, but includes the production habits that matter: timeouts, limits, retries, and clean shutdown. You can copy these patterns into your own services.

Prep: environment and dependencies

Project 1 — Async scraper with bounded concurrency

Scraping is a perfect asyncio use case: the program spends most of its time waiting on HTTP responses. The trick is not “go faster” — it’s “go fast without being rude”: limit concurrency, use timeouts, and retry intelligently.

import asyncio
from dataclasses import dataclass
from typing import Iterable, Optional

import aiohttp


@dataclass(frozen=True)
class FetchResult:
    url: str
    ok: bool
    status: int
    bytes_read: int


async def fetch_with_retry(
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    url: str,
    *,
    timeout_s: float = 10.0,
    retries: int = 2,
    backoff_base_s: float = 0.4,
) -> FetchResult:
    # Keep the event loop responsive: acquire semaphore, then await I/O.
    async with sem:
        for attempt in range(retries + 1):
            try:
                timeout = aiohttp.ClientTimeout(total=timeout_s)
                async with session.get(url, timeout=timeout) as resp:
                    body = await resp.read()
                    return FetchResult(
                        url=url,
                        ok=(200 <= resp.status < 300),
                        status=resp.status,
                        bytes_read=len(body),
                    )
            except (aiohttp.ClientError, asyncio.TimeoutError):
                if attempt >= retries:
                    return FetchResult(url=url, ok=False, status=0, bytes_read=0)
                await asyncio.sleep(backoff_base_s * (2 ** attempt))


async def scrape(urls: Iterable[str], *, concurrency: int = 20) -> list[FetchResult]:
    sem = asyncio.Semaphore(concurrency)
    headers = {"User-Agent": "UniLabAsyncScraper/1.0 (+https://bloglab-65579.firebaseapp.com/)"}
    connector = aiohttp.TCPConnector(limit=concurrency, ttl_dns_cache=300)

    async with aiohttp.ClientSession(headers=headers, connector=connector) as session:
        tasks = [asyncio.create_task(fetch_with_retry(session, sem, url)) for url in urls]
        # return_exceptions=False: fail loudly during development; add logging if needed.
        return await asyncio.gather(*tasks)


def main() -> None:
    urls = [
        "https://example.com",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/404",
    ]
    results = asyncio.run(scrape(urls, concurrency=10))
    for r in results:
        print(f"{r.url} ok={r.ok} status={r.status} bytes={r.bytes_read}")


if __name__ == "__main__":
    main()

Project 2 — WebSocket client with reconnect + graceful shutdown

WebSockets are the opposite of scraping: the connection stays open and messages stream in. That means you need long-lived tasks, heartbeats, and a reconnect loop that doesn’t spin out of control when the network is flaky.

import asyncio
import contextlib
from typing import Optional

import websockets


async def heartbeat(ws: websockets.WebSocketClientProtocol, interval_s: float) -> None:
    while True:
        await asyncio.sleep(interval_s)
        # ping() returns a Future you can await for pong if you want strictness
        await ws.ping()


async def consume(ws: websockets.WebSocketClientProtocol) -> None:
    async for msg in ws:
        # Keep message handling fast; if it becomes heavy, push to a queue/to_thread().
        print("recv:", msg)


async def run_client(url: str, *, stop: asyncio.Event) -> None:
    backoff_s = 0.5
    max_backoff_s = 10.0

    while not stop.is_set():
        try:
            async with websockets.connect(url, ping_interval=None) as ws:
                backoff_s = 0.5  # reset on successful connect
                hb = asyncio.create_task(heartbeat(ws, interval_s=15.0))
                consumer = asyncio.create_task(consume(ws))

                # Wait until stop is set OR a task fails/completes.
                done, pending = await asyncio.wait(
                    {hb, consumer, asyncio.create_task(stop.wait())},
                    return_when=asyncio.FIRST_COMPLETED,
                )

                for t in pending:
                    t.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await asyncio.gather(*pending)

        except (OSError, websockets.WebSocketException):
            await asyncio.sleep(backoff_s)
            backoff_s = min(max_backoff_s, backoff_s * 2)


async def main() -> None:
    stop = asyncio.Event()
    url = "wss://echo.websocket.events"  # public echo server (demo)

    client = asyncio.create_task(run_client(url, stop=stop))
    try:
        # Send a few demo messages by opening a short-lived connection (optional).
        async with websockets.connect(url) as ws:
            for i in range(3):
                await ws.send(f"hello {i}")
                await asyncio.sleep(0.2)
    except Exception:
        pass

    # Let the client run briefly.
    await asyncio.sleep(3.0)
    stop.set()
    await client


if __name__ == "__main__":
    asyncio.run(main())

Project 3 — Tiny task queue with worker pool + backpressure

A task queue is a great way to learn “structured async” without any external infrastructure. You’ll build a bounded asyncio.Queue, spawn a fixed number of workers, and implement shutdown correctly. This pattern shows up everywhere: ETL, background jobs, pipelines, and fan-out processing.

import asyncio
import random
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    id: int
    payload: str
    attempts_left: int = 2


SENTINEL: Job | None = None


async def handle(job: Job) -> None:
    # Simulate I/O work (replace with HTTP/DB/etc).
    await asyncio.sleep(0.05)
    # Simulate occasional transient failures.
    if random.random() < 0.15:
        raise RuntimeError("transient failure")


async def worker(name: str, q: asyncio.Queue[Job | None]) -> None:
    while True:
        job = await q.get()
        try:
            if job is SENTINEL:
                return

            try:
                await handle(job)
                print(f"[{name}] done job={job.id}")
            except Exception:
                if job.attempts_left > 0:
                    # Re-queue with one fewer attempt (simple retry).
                    await q.put(Job(job.id, job.payload, job.attempts_left - 1))
                else:
                    print(f"[{name}] failed job={job.id} (no attempts left)")
        finally:
            q.task_done()


async def main() -> None:
    q: asyncio.Queue[Job | None] = asyncio.Queue(maxsize=50)  # bounded => backpressure
    workers = [asyncio.create_task(worker(f"w{i}", q)) for i in range(5)]

    # Producer: enqueue work. If q is full, await will pause the producer (good).
    for i in range(200):
        await q.put(Job(i, payload=f"item-{i}"))

    # Wait until all jobs (including retries) are processed.
    await q.join()

    # Shut down workers with sentinels.
    for _ in workers:
        await q.put(SENTINEL)

    await asyncio.gather(*workers)


if __name__ == "__main__":
    asyncio.run(main())

Common mistakes

Most asyncio pain comes from a few repeat offenders. If your async code feels “slow” or “unreliable,” scan this list first.

FAQ

When should I use asyncio instead of threads?

Use asyncio when you have many concurrent I/O operations (HTTP calls, sockets, DB queries) and you want predictable resource usage. Threads can work too, but they cost more memory and scheduling overhead at high concurrency. For small concurrency, threads are fine.

Will asyncio make my program faster?

It will make I/O-heavy programs more throughput-efficient by keeping one process busy while requests are in flight. It won’t speed up CPU-bound work. If your bottleneck is computation, use multiprocessing or optimized native code.

What’s the difference between a coroutine and a task?

A coroutine is the thing returned by calling an async def function. A task is a scheduled coroutine managed by the event loop. You typically create tasks when you want work to progress concurrently with other coroutines.

How do I prevent “too many open connections” errors?

Bound concurrency. Use a semaphore or a worker pool. For HTTP, reuse a single session and limit the connector concurrency. Avoid creating one session/connector per request.

How do I handle timeouts and retries correctly?

Put timeouts on every external call, then retry only on transient errors (timeouts, connection resets, 5xx). Add backoff (and ideally jitter) so failures don’t cause reconnect storms. Keep retries bounded and observable via logs/metrics.

What is structured concurrency and why should I care?

Structured concurrency means task lifetimes are scoped: child tasks belong to a parent block. In Python 3.11+, asyncio.TaskGroup helps you avoid orphan tasks and makes failures/cancellation predictable. It’s one of the simplest ways to make async code feel “safe.”

Cheatsheet

A scan-fast checklist for writing asyncio code that behaves well under load.

Wrap-up

If asyncio felt abstract before, you now have three concrete patterns you can reuse: bounded concurrent HTTP, resilient long-lived connections, and queue + workers with backpressure and clean shutdown. Those three cover a surprising amount of real-world Python concurrency.

Keep this page bookmarked: Quickstart for getting started, Cheatsheet for day-to-day patterns, and the projects when you need a template.

Related posts

• Python in 60 Minutes: From Zero to Useful
• Top 25 Python Errors (And the 1-Line Fixes)
• Python Dataclasses: Cleaner Models With Less Boilerplate
• Make Python Faster: 12 Micro-Optimizations That Matter
• Publish a Python Package the Right Way (pyproject.toml + CI)
• Build a Beautiful CLI Tool: Arguments, Colors, and TUI Basics