Python Async Await Tutorial with Diagrams & Real Examples (2026)

The Story That Explains Async / Await

async / await is Python's way of writing code that pauses cleanly at slow I/O points and lets other work run during the wait. It is single-threaded, cooperative, and dramatically faster for anything that spends most of its time waiting on the network, a database, files, or timers.

The Problem — Why Sync Code Is Slow at I/O

Consider fetching three URLs. Each one takes ~1 second because the network is slow. Synchronous code stops the whole program during every wait.

The Three Building Blocks

Your First Async Function

import asyncio

async def greet(name):
    print(f"Hello {name}, starting...")
    await asyncio.sleep(1)          # pause 1s WITHOUT blocking
    print(f"Goodbye {name}")

# Calling greet() only creates a coroutine — nothing runs yet
# asyncio.run() starts the event loop and executes it
asyncio.run(greet("Mohit"))

Visual Diagram — Inside the Event Loop

Step-by-Step: What Happens on Every Tick

Running Things Concurrently — asyncio.gather

A single await on its own doesn't buy you anything — it just waits. The speedup comes from running multiple coroutines at once with asyncio.gather().

Sequential vs Concurrent — the Same Task, Two Ways

import asyncio, time

async def fetch(name, delay):
    print(f"→ start {name}")
    await asyncio.sleep(delay)      # simulate network
    print(f"✓ done  {name}")
    return f"{name}-result"

async def sequential():
    t0 = time.perf_counter()
    a = await fetch("A", 1)     # wait for A
    b = await fetch("B", 1)     # then wait for B
    c = await fetch("C", 1)     # then wait for C
    print(f"Sequential: {time.perf_counter()-t0:.2f}s")

async def concurrent():
    t0 = time.perf_counter()
    a, b, c = await asyncio.gather(
        fetch("A", 1),
        fetch("B", 1),
        fetch("C", 1),
    )
    print(f"Concurrent: {time.perf_counter()-t0:.2f}s")

asyncio.run(sequential())
asyncio.run(concurrent())

Tasks — Fire-and-Continue Concurrency

asyncio.gather() waits for a fixed set of coroutines to all finish. Sometimes you want to start a coroutine, continue your own work, and collect the result later. That's what a Task is — a coroutine wrapped in a scheduled unit of work.

import asyncio

async def slow_price_lookup(symbol):
    await asyncio.sleep(2)      # pretend this is an API call
    return {"symbol": symbol, "price": 142.35}

async def main():
    # Fire the task — it starts running NOW, in the background
    task = asyncio.create_task(slow_price_lookup("AAPL"))

    # Meanwhile, do other work — this takes ~1 second
    print("doing local calculations...")
    await asyncio.sleep(1)
    print("calculations done, now waiting on API...")

    # Now collect the result. Only ~1 more second of waiting (not 2!)
    result = await task
    print(f"Got: {result}")

asyncio.run(main())

Practical Example — Fetching Many URLs with aiohttp

This is where async really shines. aiohttp is the async equivalent of requests. Ten URLs that would take 10 seconds sequentially can be fetched in ~1 second concurrently.

# pip install aiohttp
import asyncio, aiohttp, time

URLS = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

async def fetch_url(session, url):
    async with session.get(url) as resp:
        return resp.status, len(await resp.text())

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        return await asyncio.gather(*tasks)

t0 = time.perf_counter()
results = asyncio.run(fetch_all(URLS))
print(f"Fetched {len(results)} URLs in {time.perf_counter()-t0:.2f}s")
for status, size in results:
    print(f"  status={status}  bytes={size}")

Timeouts & Cancellation

Network calls hang. Databases stall. You need a way to say "give up if this takes too long." asyncio.wait_for() and asyncio.timeout() handle this cleanly.

import asyncio

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

async def main():
    # Modern style — Python 3.11+
    try:
        async with asyncio.timeout(2):
            result = await slow_task()
    except TimeoutError:
        print("gave up after 2 seconds")

    # Classic style — works on all versions
    try:
        result = await asyncio.wait_for(slow_task(), timeout=2)
    except asyncio.TimeoutError:
        print("also gave up after 2 seconds")

asyncio.run(main())

Manually Cancelling a Task

import asyncio

async def worker(n):
    try:
        while True:
            print(f"worker {n} tick")
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        print(f"worker {n} shutting down cleanly")
        raise                        # re-raise so cancellation propagates

async def main():
    t = asyncio.create_task(worker(1))
    await asyncio.sleep(2)
    t.cancel()                     # request cancellation
    try:
        await t
    except asyncio.CancelledError:
        pass

asyncio.run(main())

Common Pitfalls (and How to Avoid Them)

async / await vs Threads vs Processes

Real-World Example — Weather Dashboard Aggregator

Let's build something concrete: fetch current weather for 5 cities from a public API in parallel, handle failures gracefully, apply a timeout, and print a formatted summary.

import asyncio, aiohttp, time

CITIES = [
    ("Mumbai",     19.0760, 72.8777),
    ("New York",   40.7128, -74.0060),
    ("London",     51.5074, -0.1278),
    ("Tokyo",      35.6762, 139.6503),
    ("Sydney",    -33.8688, 151.2093),
]

API = "https://api.open-meteo.com/v1/forecast"

async def fetch_weather(session, city, lat, lon):
    params = {"latitude": lat, "longitude": lon, "current_weather": True}
    try:
        async with asyncio.timeout(5):
            async with session.get(API, params=params) as r:
                data = await r.json()
                cw = data["current_weather"]
                return city, cw["temperature"], cw["windspeed"], None
    except Exception as e:
        return city, None, None, str(e)

async def main():
    t0 = time.perf_counter()
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *[fetch_weather(session, c, lat, lon) for c, lat, lon in CITIES]
        )

    print(f"\nWeather report — fetched in {time.perf_counter()-t0:.2f}s")
    print("-" * 50)
    for city, temp, wind, err in results:
        if err:
            print(f"{city:12s}  ERROR: {err}")
        else:
            print(f"{city:12s}  {temp:5.1f}°C   wind {wind:.1f} km/h")

asyncio.run(main())

Limiting Concurrency — Don't DDoS Your Own API

Firing 10,000 requests at once is easy in async — and a great way to get rate-limited or crash the server. Use asyncio.Semaphore to cap in-flight tasks.

import asyncio, aiohttp

async def fetch_one(sem, session, url):
    async with sem:                          # at most N in flight
        async with session.get(url) as r:
            return await r.text()

async def fetch_many(urls, max_concurrent=10):
    sem = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(
            *[fetch_one(sem, session, u) for u in urls]
        )

# Even if urls has 10,000 entries, only 10 requests run at any moment
urls = [f"https://httpbin.org/anything/{i}" for i in range(10_000)]
results = asyncio.run(fetch_many(urls, max_concurrent=20))

When to Use async / await

Modern Pattern — asyncio.TaskGroup (Python 3.11+)

TaskGroup is the recommended replacement for gather when you want clean error handling: if any child task fails, the group cancels the rest and raises a single grouped exception.

import asyncio

async def work(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"{name} exploded")
    return f"{name} ok"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(work("A", 1))
            t2 = tg.create_task(work("B", 2, fail=True))
            t3 = tg.create_task(work("C", 3))
    except* RuntimeError as eg:      # exception group syntax
        for e in eg.exceptions:
            print(f"caught: {e}")

asyncio.run(main())

Golden Rules