Python asyncio: Complete Guide with async/await, Tasks, and Patterns

Python's asyncio library lets you write concurrent I/O-bound code using async/await syntax — a single thread that can handle thousands of connections efficiently. This guide covers everything from coroutines to production patterns.

Quick Reference

What Is asyncio?

asyncio uses an event loop — a single thread that switches between tasks whenever one is waiting for I/O. This makes it ideal for network requests, database queries, file I/O, and anything that spends time waiting.

Thread 1 (event loop):
  Task A: send HTTP request → waiting... ←─────────┐
  Task B: query database   → waiting... ←───────┐  │
  Task C: read file        → got result! ←───┐  │  │
  (switch to C, then B resolves, then A resolves)

asyncio is not for CPU-bound work (use multiprocessing for that).

async/await Basics

import asyncio

async def fetch_data(url: str) -> str:
    # Simulate I/O wait (replace with real aiohttp call)
    await asyncio.sleep(1)
    return f"data from {url}"

async def main():
    result = await fetch_data("https://example.com")
    print(result)

asyncio.run(main())  # Python 3.7+ entry point

Rules:

• async def defines a coroutine function — calling it returns a coroutine object, not the result
• await suspends the current coroutine until the awaited thing completes
• You can only await inside async def
• asyncio.run() creates the event loop and runs the top-level coroutine

Coroutines, Tasks, and Futures

import asyncio

async def greet(name: str) -> str:
    await asyncio.sleep(0.1)
    return f"Hello, {name}!"

async def main():
    # Coroutine — not running yet
    coro = greet("Alice")

    # Await it directly — runs now, blocks until done
    result = await coro
    print(result)  # Hello, Alice!

    # Task — schedules coroutine on event loop, runs concurrently
    task = asyncio.create_task(greet("Bob"))
    # ... do other work here ...
    result = await task  # wait for task to finish
    print(result)  # Hello, Bob!

asyncio.run(main())

Running Concurrently with gather

asyncio.gather() runs multiple coroutines at the same time and returns all results:

import asyncio
import time

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"✓ {url}"

async def main():
    start = time.perf_counter()

    # Sequential — takes 3 seconds total
    # r1 = await fetch("a.com", 1)
    # r2 = await fetch("b.com", 1)
    # r3 = await fetch("c.com", 1)

    # Concurrent — takes ~1 second total
    results = await asyncio.gather(
        fetch("a.com", 1),
        fetch("b.com", 1),
        fetch("c.com", 1),
    )
    print(results)  # ['✓ a.com', '✓ b.com', '✓ c.com']
    print(f"Done in {time.perf_counter() - start:.2f}s")

asyncio.run(main())

Handle errors independently with return_exceptions=True:

results = await asyncio.gather(
    fetch("a.com", 1),
    fetch("bad.com", 0),   # might raise
    return_exceptions=True  # don't cancel others on failure
)
for r in results:
    if isinstance(r, Exception):
        print(f"Error: {r}")
    else:
        print(r)

create_task for Background Work

import asyncio

async def background_job(name: str):
    for i in range(3):
        await asyncio.sleep(1)
        print(f"{name}: step {i+1}")

async def main():
    # Schedule background tasks — they start immediately
    task1 = asyncio.create_task(background_job("worker-1"))
    task2 = asyncio.create_task(background_job("worker-2"))

    print("Tasks started, doing other work...")
    await asyncio.sleep(1.5)
    print("Other work done, waiting for tasks...")

    await task1
    await task2
    print("All done!")

asyncio.run(main())

Cancel a task:

task = asyncio.create_task(long_running())
await asyncio.sleep(1)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("Task was cancelled")

Timeout with wait_for

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out!")

asyncio.run(main())

asyncio.wait for Fine-Grained Control

import asyncio

async def task(n: int) -> int:
    await asyncio.sleep(n)
    return n * 2

async def main():
    tasks = {asyncio.create_task(task(i)) for i in range(1, 4)}

    # Wait until FIRST completes
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for t in done:
        print(f"First result: {t.result()}")

    # Cancel the rest
    for t in pending:
        t.cancel()

asyncio.run(main())

Async Context Managers

Use async with for resources that need async setup/teardown:

import asyncio

class AsyncDB:
    async def __aenter__(self):
        await asyncio.sleep(0.1)  # simulate connect
        print("Connected")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await asyncio.sleep(0.05)  # simulate close
        print("Disconnected")

    async def query(self, sql: str) -> list:
        await asyncio.sleep(0.1)
        return [{"result": sql}]

async def main():
    async with AsyncDB() as db:
        rows = await db.query("SELECT 1")
        print(rows)

asyncio.run(main())

Async Generators and async for

import asyncio

async def paginate(total: int, page_size: int = 10):
    """Async generator — yields pages of data lazily"""
    for offset in range(0, total, page_size):
        await asyncio.sleep(0.05)  # simulate DB query
        page = list(range(offset, min(offset + page_size, total)))
        yield page

async def main():
    async for page in paginate(35, page_size=10):
        print(f"Processing page: {page}")

asyncio.run(main())

Producer-Consumer with asyncio.Queue

import asyncio
import random

async def producer(queue: asyncio.Queue, n: int):
    for i in range(n):
        item = random.randint(1, 100)
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # sentinel to signal done

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

async def main():
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue),
    )

asyncio.run(main())

Synchronization Primitives

Lock — mutual exclusion

import asyncio

lock = asyncio.Lock()
shared_counter = 0

async def increment():
    global shared_counter
    async with lock:          # only one coroutine at a time
        val = shared_counter
        await asyncio.sleep(0)  # yield control
        shared_counter = val + 1

async def main():
    await asyncio.gather(*[increment() for _ in range(100)])
    print(shared_counter)  # 100 (correct with lock, random without)

asyncio.run(main())

Semaphore — limit concurrency

import asyncio
import aiohttp  # pip install aiohttp

sem = asyncio.Semaphore(10)  # max 10 concurrent requests

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with sem:           # blocks if 10 already running
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [f"https://httpbin.org/get?n={i}" for i in range(50)]
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch(session, u) for u in urls])
    print(f"Fetched {len(results)} pages")

asyncio.run(main())

Event — one-time signal

import asyncio

event = asyncio.Event()

async def waiter(name: str):
    print(f"{name}: waiting for event...")
    await event.wait()
    print(f"{name}: event received!")

async def setter():
    await asyncio.sleep(1)
    print("Setting event!")
    event.set()

async def main():
    await asyncio.gather(
        waiter("A"),
        waiter("B"),
        setter(),
    )

asyncio.run(main())

Running Blocking Code

asyncio is single-threaded — blocking calls freeze the event loop. Use run_in_executor to offload them:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io(path: str) -> str:
    time.sleep(1)  # simulates slow disk read
    return f"content of {path}"

def cpu_intensive(n: int) -> int:
    return sum(i * i for i in range(n))  # heavy computation

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

    # Thread pool for I/O-bound blocking calls
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io, "data.txt")
        print(result)

    # Process pool for CPU-bound work
    from concurrent.futures import ProcessPoolExecutor
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_intensive, 10_000_000)
        print(result)

asyncio.run(main())

Real-World Pattern: HTTP Fetcher with aiohttp

import asyncio
import aiohttp
from dataclasses import dataclass

@dataclass
class FetchResult:
    url: str
    status: int
    body: str

async def fetch_one(
    session: aiohttp.ClientSession,
    url: str,
    sem: asyncio.Semaphore,
    timeout: float = 10.0,
) -> FetchResult:
    async with sem:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
                body = await resp.text()
                return FetchResult(url=url, status=resp.status, body=body[:200])
        except Exception as e:
            return FetchResult(url=url, status=0, body=str(e))

async def fetch_all(urls: list[str], concurrency: int = 20) -> list[FetchResult]:
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        return await asyncio.gather(
            *[fetch_one(session, url, sem) for url in urls]
        )

async def main():
    urls = [f"https://httpbin.org/status/{code}" for code in [200, 404, 500]]
    results = await fetch_all(urls, concurrency=10)
    for r in results:
        print(f"{r.status} {r.url}")

asyncio.run(main())

Real-World Pattern: Async Database with asyncpg

import asyncio
import asyncpg  # pip install asyncpg

async def main():
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/mydb",
        min_size=5,
        max_size=20,
    )

    async with pool.acquire() as conn:
        # Parameterised query — no SQL injection risk
        rows = await conn.fetch(
            "SELECT id, name FROM users WHERE active = $1", True
        )
        for row in rows:
            print(dict(row))

    await pool.close()

asyncio.run(main())

asyncio with TypeScript Analogy

If you know JavaScript Promises:

Common Mistakes

asyncio vs Threading vs Multiprocessing

Rule of thumb:

• Thousands of network requests → asyncio + aiohttp
• Calling sync library that blocks → ThreadPoolExecutor
• Heavy computation (ML, image processing) → ProcessPoolExecutor

Frequently Asked Questions

Can I mix sync and async code?
Yes — call asyncio.run(coro) from sync code to run one async function. From inside async code, use loop.run_in_executor() for sync functions. Don't nest asyncio.run() calls.

What's the difference between gather and create_task?
create_task schedules a coroutine and returns a Task immediately. gather wraps coroutines as tasks and waits for all to finish. Use create_task when you want to start a task and do other things before awaiting it; use gather when you want to run multiple tasks and collect all results.

Why is my async code not faster than sync?
asyncio only helps for I/O-bound work. If your code is CPU-bound (computations), asyncio won't help — use multiprocessing. Also make sure you're actually using async libraries (aiohttp not requests, asyncpg not psycopg2).

How do I run asyncio in Jupyter?
Jupyter already runs an event loop, so you can't asyncio.run(). Instead, await directly in cells: result = await my_coroutine(). For scripts embedded in Jupyter, use nest_asyncio: import nest_asyncio; nest_asyncio.apply().

What is asyncio.sleep(0)?
It yields control to the event loop for one cycle without actually sleeping. Useful in tight loops to allow other tasks to run:

async def tight_loop():
    for i in range(1_000_000):
        process(i)
        if i % 1000 == 0:
            await asyncio.sleep(0)  # let other tasks run

How do I debug asyncio code?
Enable debug mode: asyncio.run(main(), debug=True) — it warns about slow callbacks, unawaited coroutines, and wrong thread usage. Also useful: PYTHONASYNCIODEBUG=1 env variable, and asyncio.all_tasks() to inspect running tasks.