What you'll learn
Quick Answer
async def creates a coroutine, and await hands control back to the event loop while something slow finishes. The loop runs one task at a time on one thread, so any blocking call (time.sleep, requests.get, a heavy loop) freezes every other task. Async helps when your program spends its time waiting on the network, a database or a disk. It does nothing for CPU-heavy work, which needs processes instead.
What await actually does
The single most common misunderstanding about asyncio is that it makes code run in parallel. It does not. A normal asyncio program runs on one thread. Nothing runs at the same time as anything else. What asyncio gives you is the ability to pause a function in the middle, go do something else, and come back later.
An async def function is not a normal function. Calling it does not run the body. It returns a coroutine object, which is a description of work that has not started yet. Something has to drive it, and that something is the event loop, usually started by asyncio.run().
import asyncio
async def fetch_marks(roll):
print("asking for", roll)
await asyncio.sleep(1) # pretend this is a network call
return {"roll": roll, "marks": 78}
# This does NOT run the function:
c = fetch_marks("CS21")
print(c) # <coroutine object fetch_marks at 0x...>
# This does:
print(asyncio.run(fetch_marks("CS21")))The word await is the interesting part. It means: this thing might take a while; if it is not ready, suspend me and let the loop run somebody else. Every await is a permission slip you hand to the scheduler. Code between two awaits runs start to finish with no interruption, which is why asyncio code rarely needs locks the way threaded code does.
The flip side is the rule that governs everything else in this article. If a stretch of code contains no await, the loop cannot take control back. Your function keeps the whole program to itself until it either finishes or hits an await. That is the entire model, and almost every asyncio bug is a violation of it.
If you forget the await entirely, Python does not crash. It builds the coroutine, throws it away, and prints RuntimeWarning: coroutine 'fetch_marks' was never awaited. Your function simply never ran, and the return value you were expecting is a coroutine object. Treat that warning as an error.
Blocking calls freeze the whole loop
Here is the failure mode that catches nearly everyone the first time. You write async code, add time.sleep(1) because it is the sleep you know, and the whole point of asyncio quietly disappears.
import asyncio, time
async def bad(n):
time.sleep(1) # blocks the thread, and so the loop
return n
async def good(n):
await asyncio.sleep(1) # yields control back to the loop
return n
async def main():
start = time.perf_counter()
await asyncio.gather(*[bad(i) for i in range(5)])
print("bad :", round(time.perf_counter() - start, 1), "s")
start = time.perf_counter()
await asyncio.gather(*[good(i) for i in range(5)])
print("good:", round(time.perf_counter() - start, 1), "s")
asyncio.run(main())The first block takes roughly five seconds, the second roughly one. time.sleep puts the operating system thread to sleep, and the event loop lives on that thread, so every other task is frozen with it. asyncio.sleep registers a timer and suspends only the current coroutine.
The same trap applies to real libraries, and it is much harder to spot there. requests.get() is blocking. So is open(...).read() on a slow disk, so is a psycopg or mysqlclient query, so is subprocess.run. Putting them inside async def changes nothing at all; they still hold the thread. Async only helps if the library itself is written to cooperate, which is why aiohttp and httpx.AsyncClient exist alongside requests.
When you are stuck with a blocking library, do not pretend. Push it onto a worker thread with asyncio.to_thread, available since Python 3.9, so the loop stays free.
import asyncio, requests
async def fetch(url):
# requests is blocking, so run it off the loop thread
return await asyncio.to_thread(requests.get, url)Running things concurrently with gather and tasks
Awaiting one coroutine after another is still sequential. Three await statements in a row take as long as the three calls added together. To overlap them you have to hand the loop several jobs at once.
import asyncio
async def check_seat(city):
await asyncio.sleep(1)
return f"{city}: 12 seats"
async def main():
# sequential, about 3 seconds
a = await check_seat("Pune")
b = await check_seat("Chennai")
c = await check_seat("Jaipur")
# concurrent, about 1 second
results = await asyncio.gather(
check_seat("Pune"),
check_seat("Chennai"),
check_seat("Jaipur"),
)
print(results)
asyncio.run(main())asyncio.gather returns results in the order you passed them in, not the order they finished, which is usually what you want when you are zipping results back onto a list of inputs. By default, if one coroutine raises, gather re-raises immediately and the results of the others are lost. Pass return_exceptions=True to get a list where failures appear as exception objects instead, then filter them yourself.
asyncio.create_task is the other tool. It schedules a coroutine right away and gives you a handle you can await later, which is useful when you want work to start before you need the answer. There is a sharp edge here: the loop keeps only a weak reference to a task, so a task nobody holds a reference to can be garbage collected mid-flight and simply vanish. If you fire and forget, keep the handle in a set.
background = set()
def spawn(coro):
task = asyncio.create_task(coro)
background.add(task)
task.add_done_callback(background.discard)
return taskOne more practical point: gather over ten thousand URLs will genuinely open ten thousand connections and get you rate limited or blocked. Put a asyncio.Semaphore(20) around the work and async with it inside each task to cap how many run at once.
I/O bound helps, CPU bound does not
Ask one question before reaching for asyncio: is my program slow because it is waiting, or because it is calculating? Async only helps with waiting.
Waiting means network requests, database round trips, reading files, calling a payment gateway, talking to Redis. During that wait the CPU is idle, and asyncio spends that idle time on another task. This is why an API server handling hundreds of slow database calls benefits enormously from async, even on a single core.
Calculating means resizing images, parsing a huge CSV, hashing passwords, running a sorting or dynamic programming routine, training anything. There is no waiting to reclaim. The CPU is already busy, and because CPython holds the global interpreter lock while running Python bytecode, threads will not save you either.
import asyncio
from concurrent.futures import ProcessPoolExecutor
def heavy(n):
return sum(i * i for i in range(n))
async def wrong(n):
return heavy(n) # blocks the loop for the full duration
async def right(pool, n):
loop = asyncio.get_running_loop()
# a separate process has its own interpreter, so its own GIL
return await loop.run_in_executor(pool, heavy, n)
async def main():
with ProcessPoolExecutor() as pool:
jobs = [right(pool, 5_000_000) for _ in range(4)]
print(await asyncio.gather(*jobs))
if __name__ == "__main__": # not optional on Windows or macOS
asyncio.run(main())Separate processes mean separate interpreters and separate GILs, so the work genuinely runs on several cores. The __main__ guard is required wherever child processes are started by spawning rather than forking, which includes Windows and modern macOS, because each child re-imports your module. The cost is that arguments and return values must be picklable and get copied between processes, so it only pays off when each job is chunky.
A quick decision rule for placement interviews and for real work: multiple slow network calls, use asyncio. A blocking library you cannot replace, use threads. Heavy computation, use processes. Mixing these up is the reason so many people report that async made no difference at all.
Mistakes that cost hours
Calling asyncio.run inside a running loop. In a Jupyter notebook, or inside an already-async web handler, asyncio.run() raises RuntimeError: asyncio.run() cannot be called from a running event loop. Inside async code you just await the coroutine. asyncio.run belongs at the top level of your program, once.
Awaiting inside a normal def. await is a syntax error outside async def. If you need async work from sync code, the boundary has to be crossed with asyncio.run at the entry point, not sprinkled in the middle.
Using a blocking client by accident. Import lists are worth auditing. If your async view imports requests, time.sleep, or a synchronous ORM, the server is not concurrent no matter how many async def keywords it contains. Under load this shows up as request latency that rises together for every user at once.
Sharing one client badly. Creating a new httpx.AsyncClient per request throws away connection pooling. Create one and reuse it, closing it on shutdown with async with or a lifespan hook.
Ignoring cancellation. When a client disconnects or a timeout fires, your task receives asyncio.CancelledError at its next await point. Swallowing it with a bare except Exception is usually wrong; in modern Python CancelledError inherits from BaseException so a plain except Exception will not catch it, but a bare except: will and that leaves zombie tasks running.
import asyncio
async def main():
try:
await asyncio.wait_for(slow_call(), timeout=5)
except asyncio.TimeoutError:
print("gateway did not answer in 5s")Finally, debug mode is free and underused. Run with PYTHONASYNCIODEBUG=1 and Python will warn you when a coroutine blocks the loop for too long, which is the fastest way to find the one stray blocking call hiding in a large codebase.