One Thread, Many Tasks: A Practical Deep Dive into Python's asyncio

Why asyncio exists

Most programs spend the majority of their time waiting: waiting for an HTTP response, a database query, a file read, a socket. While one request sits idle, your CPU is doing nothing useful. The traditional fix is threads, but threads carry overhead and force you to reason about locks and race conditions. asyncio offers a different model: a single thread that juggles thousands of waiting operations by switching between them at well-defined suspension points. When one task pauses to wait for I/O, the event loop runs another. Nothing runs in true parallel, but nothing sits blocked either.

The key mental model: asyncio gives you concurrency, not parallelism. It shines for I/O-bound workloads (web scraping, API clients, servers, network tools) and does nothing for CPU-bound work. Keep that distinction in mind and the rest falls into place.

Coroutines, await, and the event loop

A function defined with async def is a coroutine function. Calling it does not run it — it returns a coroutine object. To actually execute it you either await it from inside another coroutine, or hand it to asyncio.run(), which starts an event loop, runs the coroutine to completion, and shuts the loop down.

import asyncio

async def greet(name):
    await asyncio.sleep(1)   # simulates awaiting I/O
    return f"Hello, {name}"

async def main():
    message = await greet("world")
    print(message)

asyncio.run(main())

The await keyword is the heart of it. It means "suspend here until this finishes, and let the event loop do other things in the meantime." You can only use await inside an async def. Note that asyncio.sleep() is the async cousin of time.sleep() — the former yields control back to the loop, the latter blocks the whole thread. Mixing up the two is the single most common beginner mistake.

Running things concurrently with gather

Awaiting coroutines one after another is sequential — you gain nothing. The payoff comes when you run several at once. asyncio.gather() schedules multiple awaitables concurrently and waits for all of them, returning their results in order.

import asyncio, time

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done in {delay}s"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("A", 0.2),
        fetch("B", 0.1),
        fetch("C", 0.3),
    )
    print(results)
    print(f"Elapsed ~{time.perf_counter() - start:.2f}s")

asyncio.run(main())
# ['A done in 0.2s', 'B done in 0.1s', 'C done in 0.3s']
# Elapsed ~0.30s   <- not 0.6s, because they overlap

Three tasks that would take 0.6 seconds run in about 0.3, the length of the slowest one. That overlap is the whole point of asyncio.

By default, if any coroutine in a gather raises, the exception propagates immediately and the others may be left running. Pass return_exceptions=True to collect exceptions as ordinary results instead of aborting:

async def boom():
    raise ValueError("nope")

async def fine():
    await asyncio.sleep(0.01)
    return 1

async def main():
    results = await asyncio.gather(boom(), fine(), return_exceptions=True)
    print(results)   # [ValueError('nope'), 1]

asyncio.run(main())

Tasks: scheduling work in the background

When you await a coroutine directly, you block until it returns. Sometimes you want to start work and keep going. asyncio.create_task() wraps a coroutine in a Task and schedules it on the loop right away, so it begins running the next time you hit an await.

async def worker(n):
    await asyncio.sleep(0.05 * n)
    return n * n

async def main():
    # All five tasks start immediately, in the background.
    tasks = [asyncio.create_task(worker(i)) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(results)   # [0, 1, 4, 9, 16]

asyncio.run(main())

A subtle but important rule: keep a reference to every task you create. The event loop only holds a weak reference, so a task with no strong reference can be garbage-collected mid-flight and silently vanish. Storing tasks in a list, as above, is enough.

TaskGroup: structured concurrency (Python 3.11+)

Managing a bag of loose tasks and remembering to await them is error-prone. Python 3.11 introduced asyncio.TaskGroup, a context manager that owns its tasks and guarantees they all finish before the block exits. If any task raises, the group cancels the rest and re-raises the error (bundled in an ExceptionGroup). This "structured concurrency" pattern is now the recommended default over bare gather for anything non-trivial.

async def fetch(name):
    await asyncio.sleep(0.1)
    return name.upper()

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("alice"))
        t2 = tg.create_task(fetch("bob"))
    # On exiting the block, both tasks are guaranteed complete.
    print(t1.result(), t2.result())   # ALICE BOB

asyncio.run(main())

Timeouts and cancellation

Any real network code needs a timeout. asyncio.wait_for() wraps an awaitable and cancels it if it runs too long, raising asyncio.TimeoutError.

async def slow():
    await asyncio.sleep(5)

async def main():
    try:
        await asyncio.wait_for(slow(), timeout=0.1)
    except asyncio.TimeoutError:
        print("timed out — moving on")

asyncio.run(main())

Cancellation in asyncio works by injecting a CancelledError into the coroutine at its next suspension point. If you need cleanup, catch it, do your teardown, and then re-raise — swallowing CancelledError breaks the cancellation contract and can hang your program. On Python 3.11+, asyncio.timeout() offers a cleaner context-manager form of the same idea.

Limiting concurrency with a semaphore

Firing 10,000 requests at once will get you rate-limited or run you out of sockets. An asyncio.Semaphore caps how many coroutines run a protected section simultaneously.

async def download(sem, url):
    async with sem:                 # at most N inside here at once
        await asyncio.sleep(0.05)   # pretend to fetch
        return f"got {url}"

async def main():
    sem = asyncio.Semaphore(2)      # never more than 2 concurrent
    urls = [f"/page/{i}" for i in range(6)]
    results = await asyncio.gather(*(download(sem, u) for u in urls))
    print(results)

asyncio.run(main())

Async iterators and generators

You can stream values over time with an async def that yields, then consume it with async for. This is perfect for paginated APIs or long-lived streams where each item may require awaiting.

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(0.02)
        yield i

async def main():
    values = [x async for x in ticker(4)]  # async comprehension
    print(values)   # [0, 1, 2, 3]

asyncio.run(main())

Bridging to blocking code with to_thread

Sooner or later you will need to call a library that only offers a blocking, synchronous API. Calling it directly inside a coroutine freezes the entire event loop. The escape hatch is asyncio.to_thread(), which runs the blocking call in a worker thread and gives you back an awaitable.

import asyncio, time

def blocking_io():
    time.sleep(0.1)      # a synchronous library call
    return "result"

async def main():
    result = await asyncio.to_thread(blocking_io)
    print(result)

asyncio.run(main())

Use this for I/O-bound blocking calls. For CPU-bound work, reach for concurrent.futures.ProcessPoolExecutor via loop.run_in_executor() instead — threads won't help there because of the GIL.

Common pitfalls

A few traps catch nearly everyone:

• Calling time.sleep() in a coroutine. It blocks the whole loop. Always use await asyncio.sleep() for delays.
• Forgetting to await. Writing greet("x") without await creates a coroutine object that never runs and emits a "coroutine was never awaited" warning.
• Blocking libraries. A synchronous requests.get() inside asyncio stalls everything. Use an async client like httpx.AsyncClient or wrap the call in asyncio.to_thread().
• Expecting a speedup for CPU work. Hashing, parsing, or number-crunching won't get faster; there is still only one thread doing the computing.
• Dropping task references. Store tasks you create so the garbage collector doesn't reap them mid-run.

Wrap-up and next steps

asyncio boils down to a handful of ideas: coroutines you define with async def, suspension points marked by await, an event loop started by asyncio.run(), and concurrency you unlock with gather, create_task, or a TaskGroup. Layer on timeouts, semaphores, and to_thread and you can write clean, high-throughput I/O code without touching a single lock.

From here, build something real. Point httpx.AsyncClient at a batch of URLs and fetch them concurrently behind a semaphore. Explore asyncio.Queue for producer/consumer pipelines. If you are on Python 3.11 or newer, make TaskGroup and asyncio.timeout() your defaults — structured concurrency will save you from a whole category of bugs. The moment your program spends its life waiting on the network, asyncio is how you make that wait productive.