Introduction

asyncio is Python's built-in library for writing asynchronous, concurrent code using the async/await syntax. It provides an event loop that runs many coroutines on a single thread, letting your program handle thousands of I/O operations — network calls, file access, database queries — without blocking or spinning up expensive threads.

Real-World Analogy

Prerequisites

Core Concepts of asyncio

Your First asyncio Program

Every asyncio program starts with an async def coroutine, launched by asyncio.run().

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)   # non-blocking pause for 1 second
    print("World")

asyncio.run(main())          # creates the event loop and runs main()

Running Tasks Concurrently

The real power of asyncio appears when you run multiple coroutines at once.

Sequential (slow)

import asyncio

async def task(name, seconds):
    print(f"Start {name}")
    await asyncio.sleep(seconds)
    print(f"Done {name}")

async def main():
    await task("A", 2)   # waits 2s
    await task("B", 2)   # then another 2s  -> total ~4s

asyncio.run(main())

Concurrent with gather (fast)

import asyncio

async def task(name, seconds):
    print(f"Start {name}")
    await asyncio.sleep(seconds)
    print(f"Done {name}")
    return name

async def main():
    # Both run at the same time -> total ~2s, not 4s
    results = await asyncio.gather(
        task("A", 2),
        task("B", 2),
    )
    print(results)

asyncio.run(main())

Using create_task

import asyncio

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

async def main():
    # Schedule tasks to start immediately
    t1 = asyncio.create_task(worker(1))
    t2 = asyncio.create_task(worker(2))

    # Do other work here if needed...

    print(await t1, await t2)   # collect results

asyncio.run(main())

Essential asyncio Functions

Real Example: Concurrent HTTP Requests

Fetching many URLs concurrently is the classic asyncio use case.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        data = await response.text()
        print(f"{url} -> {len(data)} bytes")
        return data

async def main():
    urls = [
        "https://example.com",
        "https://python.org",
        "https://github.com",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        await asyncio.gather(*tasks)   # all three run concurrently

asyncio.run(main())

Timeouts and Cancellation

Protect against operations that hang using wait_for.

import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return "finished"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=3)
        print(result)
    except asyncio.TimeoutError:
        print("Task took too long and was cancelled")

asyncio.run(main())

The Performance Win

For \(N\) I/O tasks each taking time \(t\), running them sequentially costs:

\[ T_{\text{sequential}} = N \times t \]

Running them concurrently with asyncio.gather, the total is close to the longest single task:

\[ T_{\text{gather}} \approx \max_{1 \le i \le N} t_i \]

So 100 requests of 1 second each drop from ~100 seconds to roughly ~1 second — that's the asyncio advantage.

asyncio vs Threading vs Multiprocessing

Best Practices

Common Mistakes

Interview Questions

Quick Revision