Python's async/await: How It Actually Works and Why You Need It

When I joined the IdleRPG Discord bot project, it was running on Heroku with three servers and a fully synchronous Python codebase. Every database call blocked the entire process. Every network request froze every other user until it finished. The bot would lock up for seconds at a time under any load.

The developer needed help transitioning to async/await. I had been ignoring Python for years, coming from PHP and JavaScript. But I took it on. That bot went from three servers to 100,000+ servers on a single dedicated machine. The async rewrite was most of why.

I've since seen the same synchronous bottleneck in Django codebases, Flask APIs, and data pipelines. The mistake is always the same: treating I/O as if it's free. It isn't. And asyncio is how Python lets you stop paying for it.

This article covers the event loop model before any code, then builds from coroutines up to production patterns for concurrent I/O. The goal isn't a reference – the Python docs handle that. The goal is the mental model that makes all of it make sense.

The Problem asyncio Solves

Python has a Global Interpreter Lock (GIL). Only one thread executes Python bytecode at a time. Threads exist, but for CPU-bound work they don't give you parallelism. The GIL serializes them.

For I/O-bound work, threads can help -- when one thread is blocked waiting on a network response, the GIL releases and another thread can run. But threads carry overhead. Each one needs a stack. Context switches are managed by the OS. And you can't spin up 10,000 threads to handle 10,000 concurrent connections.

asyncio takes a different approach: cooperative multitasking on a single thread. Your code explicitly yields control at suspension points (the await keyword). The event loop catches that yield, runs other work, and comes back when the awaited operation finishes. One thread. No OS context switches. 10,000 concurrent connections on a Discord bot that cost $80/month to host.

That's the pitch. Now the mechanics.

The Event Loop Is a While Loop

Before writing a single line of asyncio code, understand this: the event loop is a while True loop that checks a queue of work. When you await something, you're telling the loop "I'm paused, go do something else, come back when this is ready."

Stripped to its core, it looks like this:

# Conceptual model -- not the actual asyncio source
# Source: simplified from CPython's Lib/asyncio/base_events.py
# https://github.com/python/cpython/blob/main/Lib/asyncio/base_events.py
 
ready = []          # callbacks ready to run now
scheduled = []      # callbacks waiting for a future time
io_callbacks = {}   # file descriptors -> callbacks
 
while True:
    # 1. Run everything that's ready
    for callback in ready:
        callback()
 
    # 2. Check I/O -- block for the timeout duration
    timeout = compute_timeout(scheduled)
    events = selector.select(timeout)
 
    # 3. Move completed I/O callbacks into ready
    for fd, event in events:
        io_callbacks[fd]()
 
    # 4. Move any scheduled callbacks whose time has come into ready
    for handle in scheduled:
        if handle._when <= loop.time():
            ready.append(handle)

selector.select() is a system call -- epoll on Linux, kqueue on macOS. It blocks until at least one file descriptor is ready or the timeout expires. This is how asyncio handles thousands of concurrent connections without threads: the OS watches all the sockets, and select() returns when any of them have data. You don't need a thread per connection; you need one thread watching all of them.

When your coroutine does await asyncio.sleep(5), it registers a callback in scheduled for five seconds from now and yields. The loop moves on. When the five seconds are up, your coroutine gets moved back into ready and resumes from where it paused.

This is cooperative multitasking. The "cooperative" part means the event loop trusts your coroutines to yield regularly. If you write a coroutine that does ten seconds of CPU work without an await, you block the entire event loop for ten seconds. Every other connection waits. The Discord bot would time out and get disconnected from the gateway. This matters and we'll come back to it.

What async def Actually Creates

A function defined with async def is not a function that runs asynchronously. It's a function that returns a coroutine object when called. The coroutine doesn't execute until you run it on an event loop.

# Illustrating the difference between calling and awaiting a coroutine
 
async def fetch():
    return "Hello World"
 
# Calling it returns a coroutine object -- nothing executes
result = fetch()
print(result)
# <coroutine object fetch at 0x7f3a1b2c3d40>
# Python will also warn you: RuntimeWarning: coroutine 'fetch' was never awaited
 
# To get the return value, await it inside another coroutine
async def main():
    result = await fetch()
    print(result)
    # Hello World
 
# To run the top-level coroutine, use asyncio.run()
import asyncio
asyncio.run(main())

asyncio.run() is the entry point for asyncio code in Python 3.7+. It creates an event loop, runs your coroutine until it completes, and closes the loop. You call it once, at the top level. You don't call it inside another coroutine -- that's a common mistake and it raises a RuntimeError because you'd be trying to create a second event loop inside the existing one.

await does two things. It unwraps the value from an awaitable object (a coroutine, a Task, or a Future). And it yields control back to the event loop, which can run other work while the awaitable hasn't completed yet. The second part is what makes concurrency possible.

The RuntimeWarning: coroutine was never awaited warning matters. If you see it in production logs, you have a bug – somewhere a coroutine got created and discarded without executing. This means work silently didn't happen.

gather() and create_task()

This is the part the original article completely missed. Running one coroutine at a time with await is not concurrent. It's sequential with extra steps. To get actual concurrency, you need asyncio.gather() or asyncio.create_task().

Here's the difference between sequential and concurrent execution -- and the timing gap makes this obvious:

import asyncio
import time
 
# Source: pattern adapted from discord.py's connection handling
# https://github.com/Rapptz/discord.py/blob/master/discord/gateway.py
 
async def fetch_user_data(user_id: int) -> dict:
    """Simulate a database query that takes 1 second"""
    await asyncio.sleep(1)
    return {"id": user_id, "name": f"User {user_id}"}
 
async def sequential():
    """This takes 3 seconds -- each query waits for the previous one"""
    start = time.perf_counter()
 
    user1 = await fetch_user_data(1)
    user2 = await fetch_user_data(2)
    user3 = await fetch_user_data(3)
 
    elapsed = time.perf_counter() - start
    print(f"Sequential: {elapsed:.2f}s")   # ~3.00s
    return [user1, user2, user3]
 
async def concurrent():
    """This takes 1 second -- all three queries run at the same time"""
    start = time.perf_counter()
 
    results = await asyncio.gather(
        fetch_user_data(1),
        fetch_user_data(2),
        fetch_user_data(3),
    )
 
    elapsed = time.perf_counter() - start
    print(f"Concurrent: {elapsed:.2f}s")   # ~1.00s
    return results
 
asyncio.run(concurrent())

asyncio.gather() takes multiple awaitables and schedules all of them as tasks. The event loop interleaves them -- when user1's query yields at await asyncio.sleep(1), the loop starts user2. When user2 yields, it starts user3. All three are in flight simultaneously. The total time is the duration of the slowest one, not the sum of all of them.

For 10,000 concurrent Discord gateway connections, this difference between sequential and concurrent isn't academic -- it's the difference between a bot that works and one that gets rate-limited and kicked.

asyncio.create_task() gives you more control. It schedules a coroutine as a background task and returns a Task object immediately. The task runs concurrently from the moment you create it:

import asyncio
 
async def background_cache_refresh(user_id: int):
    """Fire-and-forget cache update -- doesn't block the caller"""
    await asyncio.sleep(0.1)   # simulate cache write
    print(f"Cache refreshed for user {user_id}")
 
async def handle_command(user_id: int):
    # Start the cache refresh in the background -- we don't await it yet
    cache_task = asyncio.create_task(background_cache_refresh(user_id))
 
    # Do the main work while the cache refresh runs concurrently
    await asyncio.sleep(0.5)   # simulate main processing
    print(f"Command handled for user {user_id}")
 
    # If you need the result or want to catch exceptions, await it here
    await cache_task
 
asyncio.run(handle_command(42))
# Cache refreshed for user 42   (after ~0.1s)
# Command handled for user 42   (after ~0.5s)

One critical detail: if you create a task and don't await it or keep a reference to it, the garbage collector can destroy it mid-execution. asyncio will log a warning. Store your tasks somewhere or use asyncio.TaskGroup (below).

The Modern Pattern (Python 3.11+)

Python 3.11 added asyncio.TaskGroup, which is the recommended way to manage multiple concurrent tasks today. It handles cancellation properly -- if one task raises an exception, it cancels the others and re-raises via ExceptionGroup.

import asyncio
 
async def fetch_user_data(user_id: int) -> dict:
    await asyncio.sleep(1)
    return {"id": user_id, "name": f"User {user_id}"}
 
async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    results = []
 
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user_data(uid)) for uid in user_ids]
    
    # All tasks are complete by the time we exit the context manager
    return [task.result() for task in tasks]
 
async def main():
    users = await fetch_all_users([1, 2, 3, 4, 5])
    print(f"Fetched {len(users)} users concurrently")
 
asyncio.run(main())

The context manager guarantees that when you exit the async with block, all tasks have either completed or been cancelled. Exceptions get collected into an ExceptionGroup, which you can catch with except*. Before 3.11, you'd use asyncio.gather(return_exceptions=True) and handle it manually. TaskGroup is cleaner.

Concurrent HTTP Requests with aiohttp

requests is synchronous. Using it in an asyncio application blocks the event loop on every HTTP call. For a Discord bot making API calls or a web scraper, this kills concurrency entirely. The async replacement is aiohttp.

import asyncio
import aiohttp
 
# Pattern from aiohttp's own documentation and real production usage
# https://github.com/aio-libs/aiohttp/blob/master/docs/client_quickstart.rst
 
URLS = [
    "https://api.github.com/repos/python/cpython",
    "https://api.github.com/repos/aio-libs/aiohttp",
    "https://api.github.com/repos/Rapptz/discord.py",
]
 
async def fetch_repo(session: aiohttp.ClientSession, url: str) -> dict:
    """Fetch a single GitHub repo -- one network round trip"""
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.json()
 
async def fetch_all_repos(urls: list[str]) -> list[dict]:
    # One session for all requests -- reuses TCP connections
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_repo(session, url) for url in urls]
        return await asyncio.gather(*tasks)
 
async def main():
    repos = await fetch_all_repos(URLS)
    for repo in repos:
        print(f"{repo['full_name']}: {repo['stargazers_count']} stars")
 
asyncio.run(main())

A few things to note here.

The aiohttp.ClientSession is the async equivalent of a requests.Session. Create it once and reuse it. Creating a new session per request is expensive -- you lose connection pooling and you waste time on TLS handshakes. The async with block ensures it's closed properly when you're done.

async with session.get(url) as response gives you an async context manager. The response headers arrive first; the body is streamed separately. await response.json() waits for the full body and decodes it. If you forget the await, you get a coroutine object, not the parsed JSON. The code compiles fine. The bug is silent.

asyncio.gather(*tasks) fires all three requests concurrently. If GitHub takes 200ms to respond to each request, the total time is ~200ms, not ~600ms. At 100 URLs, the difference is ~200ms vs. ~20 seconds.

Timeouts: Don't Skip These

A coroutine that never completes blocks a task slot forever. In production, network requests can hang -- the connection succeeds, the server accepts it, and then nothing comes back. You need timeouts.

import asyncio
import aiohttp
 
async def fetch_with_timeout(
    session: aiohttp.ClientSession,
    url: str,
    timeout_seconds: float = 5.0,
) -> dict | None:
    """Fetch a URL, return None if it times out or errors"""
    try:
        async with asyncio.timeout(timeout_seconds):
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()
    except TimeoutError:
        print(f"Timeout fetching {url}")
        return None
    except aiohttp.ClientError as e:
        print(f"Request failed for {url}: {e}")
        return None

asyncio.timeout() is available in Python 3.11+. For older versions, use asyncio.wait_for(coro, timeout=5.0) which has the same semantics. Both raise TimeoutError when the deadline passes. Catch it. If you don't, the exception propagates up and can cancel parent tasks in ways you don't expect.

Note that asyncio.timeout() cancels the coroutine inside it -- it sends a cancellation signal, which raises asyncio.CancelledError inside the running coroutine, which asyncio.timeout() catches and converts to TimeoutError. If you're writing a coroutine that should be cancellation-safe, check for CancelledError in your cleanup code and re-raise it after cleanup. Never suppress CancelledError silently.

Error Handling in Coroutines

Standard try/except works inside coroutines exactly as it does in synchronous code. No special syntax.

import asyncio
import aiohttp
 
async def safe_fetch(session: aiohttp.ClientSession, url: str) -> dict | None:
    try:
        async with session.get(url) as response:
            response.raise_for_status()
            data = await response.json()
            return data
    except aiohttp.ClientResponseError as e:
        # HTTP 4xx/5xx responses
        print(f"HTTP error {e.status} for {url}")
        return None
    except aiohttp.ClientConnectionError:
        # DNS failures, refused connections
        print(f"Connection failed for {url}")
        return None
    except asyncio.CancelledError:
        # Task was cancelled externally -- re-raise, don't suppress
        print(f"Request to {url} was cancelled")
        raise

asyncio.gather() has a return_exceptions=True flag that changes how it handles errors. Without it (the default), the first exception raised by any task cancels the rest and propagates immediately. With return_exceptions=True, exceptions get returned as values in the results list, and all tasks run to completion regardless. Choose based on whether partial results are useful to you.

async def main():
    urls = ["https://valid.example.com", "https://invalid.404.example"]
 
    async with aiohttp.ClientSession() as session:
        tasks = [safe_fetch(session, url) for url in urls]
        
        # All tasks run; exceptions returned as values
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"{url} failed: {result}")
            elif result is None:
                print(f"{url} returned no data")
            else:
                print(f"{url} succeeded")

The Mistakes That Will Ruin Your Day

Blocking the event loop with synchronous code. This is the most common mistake and it's invisible until it causes problems.

import asyncio
import time
import requests  # <-- synchronous
 
async def bad_fetch(url: str):
    # This blocks the ENTIRE event loop for the duration of the request.
    # Every other coroutine waits. Your bot times out. Your API stops responding.
    response = requests.get(url)
    return response.json()
 
async def also_bad():
    # time.sleep() blocks the thread entirely -- the event loop cannot run
    time.sleep(5)
    
    # Use this instead:
    await asyncio.sleep(5)

Common synchronous blocking code that shows up in async apps: requests, psycopg2 (use asyncpg or psycopg3 in async mode instead), time.sleep(), any blocking file I/O without asyncio.to_thread(), and any heavy CPU computation.

If you have to call synchronous blocking code from an async context, run it in a thread pool executor with asyncio.to_thread():

import asyncio
import requests
 
def sync_fetch(url: str) -> dict:
    """Genuinely synchronous code that can't be replaced with aiohttp"""
    return requests.get(url).json()
 
async def async_fetch(url: str) -> dict:
    # Runs sync_fetch in a thread -- event loop is free while it executes
    return await asyncio.to_thread(sync_fetch, url)

asyncio.to_thread() submits the callable to the default ThreadPoolExecutor. The event loop remains free while the thread runs. You get the result back as an awaitable. Use this for legacy synchronous libraries you can't replace, or for file I/O where aiofiles isn't worth the dependency.

Forgetting await. Calling a coroutine without await creates the coroutine object and immediately discards it. Python warns you with RuntimeWarning: coroutine was never awaited -- but only if the coroutine object gets garbage collected before it runs. If it stays referenced somewhere, you get no warning and the work silently never happens.

async def save_to_db(data):
    await asyncio.sleep(0.1)  # simulate DB write
    print("Saved!")
 
async def main():
    save_to_db({"key": "value"})  # Bug: forgot await, nothing was saved
    await save_to_db({"key": "value"})  # Correct

Using asyncio for CPU-bound work. asyncio is for I/O concurrency. If your coroutines are doing CPU-heavy computation, they block the event loop regardless of async def. The GIL doesn't help you here, and neither does asyncio. CPU-bound work belongs in multiprocessing, not asyncio. concurrent.futures.ProcessPoolExecutor with asyncio.get_event_loop().run_in_executor() is the bridge if you need both.

A Realistic Production Pattern

Discord bots are a good reference point because the concurrency requirements are extreme. Tens of thousands of concurrent connections, each requiring real-time message processing with database reads and writes on every command.

# Source: adapted from real discord.py bot patterns
# https://github.com/Rapptz/discord.py/blob/master/examples/
 
import asyncio
import asyncpg  # async PostgreSQL driver -- psycopg2 will block your event loop
import aiohttp
from dataclasses import dataclass
 
@dataclass
class BotContext:
    db: asyncpg.Pool
    http: aiohttp.ClientSession
 
async def get_player_profile(ctx: BotContext, user_id: int) -> dict | None:
    """Fetch player data from DB -- this is I/O-bound, asyncio handles it well"""
    row = await ctx.db.fetchrow(
        "SELECT * FROM players WHERE user_id = $1",
        user_id,
    )
    return dict(row) if row else None
 
async def update_player_and_log(
    ctx: BotContext,
    user_id: int,
    new_xp: int,
) -> None:
    """
    Update player XP and log to an external API concurrently.
    Both happen at the same time -- neither waits for the other.
    """
    async with asyncio.TaskGroup() as tg:
        db_task = tg.create_task(
            ctx.db.execute(
                "UPDATE players SET xp = xp + $1 WHERE user_id = $2",
                new_xp,
                user_id,
            )
        )
        log_task = tg.create_task(
            ctx.http.post(
                "https://logging.example.com/events",
                json={"user_id": user_id, "xp_gained": new_xp},
            )
        )
    # Both tasks complete before we continue
 
async def setup() -> BotContext:
    """Create the shared connection pool and HTTP session at startup"""
    db = await asyncpg.create_pool(
        "postgresql://user:password@localhost/botdb",
        min_size=5,
        max_size=20,
    )
    http = aiohttp.ClientSession()
    return BotContext(db=db, http=http)
 
async def main():
    ctx = await setup()
    
    try:
        profile = await get_player_profile(ctx, user_id=12345)
        if profile:
            await update_player_and_log(ctx, user_id=12345, new_xp=100)
    finally:
        await ctx.db.close()
        await ctx.http.close()
 
asyncio.run(main())

A few production decisions visible in this code:

asyncpg.create_pool() creates a pool of database connections rather than a single connection. Multiple coroutines can use the pool concurrently -- each borrows a connection, runs the query, and returns it. A single connection would serialize all DB access. The pool is sized based on expected concurrency and your database's max_connections setting.

aiohttp.ClientSession is shared across all coroutines for the same reason. One session. Connection pooling. TLS sessions cached.

The TaskGroup in update_player_and_log runs the DB write and the logging HTTP call concurrently. If the logging call takes 500ms but the DB write takes 50ms, the function takes ~500ms, not ~550ms. At scale, that 50ms adds up.

When async/await Is the Wrong Tool

asyncio excels at I/O concurrency. It's the wrong tool for CPU-bound parallelism.

If you're processing images, training a model, running numerical simulations, or doing any work that saturates a CPU core, asyncio adds overhead without helping. The GIL means your CPU-bound coroutines don't run truly in parallel. They just cooperatively yield to each other and share the same single thread.

For CPU-bound parallelism, use multiprocessing.Pool or concurrent.futures.ProcessPoolExecutor. Each process gets its own GIL and its own CPU core. If you're building a pipeline that mixes async I/O with CPU processing, bridge them with loop.run_in_executor(executor, fn, *args), where executor is a ProcessPoolExecutor.

The distinction is important: async for I/O, processes for CPU.

Connecting the Dots

The Discord bot I worked on handled 100,000+ concurrent server connections. The async/await rewrite was the enabling change. Not hardware. Not caching. The architectural shift from "block on every I/O call" to "yield and let something else run."

If you want to see how Python's own interpreter handles execution flow at a lower level, the bytecode VM deep dive is a good companion read -- it explains the execution model that asyncio sits on top of.

For web development, both Django's async views (available since Django 3.1) and Flask's async support (added in Flask 2.0) use the same asyncio model described here. The same mental model applies.

And if you're thinking about Python in WebAssembly environments, threading support in WASI 0.3 will eventually unlock asyncio's thread executor in Wasm builds -- something that currently doesn't work because WASI 0.1 has no threading. The event loop itself runs fine. The asyncio.to_thread() escape hatch doesn't.

The event loop is a while loop checking a queue. Your await points are the moments you tell it to check. Keep them frequent, keep your coroutines non-blocking, and asyncio will handle the concurrency.

Python's asyncio documentation is at docs.python.org/3/library/asyncio.html. For the aiohttp client library, see docs.aiohttp.org. For async PostgreSQL, magicstack/asyncpg is the library to use.