Async Programming in Python: async, await & asyncio

Learn async/await syntax, the event loop, concurrent tasks with asyncio.gather, and why FastAPI uses async for high-performance I/O.

Modern web applications routinely wait — for database queries to return, for external API responses, for files to be read from disk. In traditional synchronous code, each waiting operation blocks the entire thread, preventing your server from handling other requests. Python’s asyncio framework with async/await syntax solves this problem by letting a single thread manage thousands of concurrent I/O operations, handing control to the event loop while one task waits so it can immediately start working on another. This is exactly how FastAPI achieves high throughput without requiring multiple threads or processes.

Concurrency vs. Parallelism

Before writing async code, it’s important to distinguish between two related but different concepts:

	Concurrency	Parallelism
What it does	Handles multiple tasks by switching between them during idle waits	Performs multiple tasks at the exact same time on multiple CPU cores
Bound by	I/O (network, disk, database)	CPU (computation, rendering, encoding)
Python tool	asyncio with async/await	multiprocessing module
FastAPI use case	✅ Handling thousands of web requests	Rare in typical web APIs

FastAPI is built for I/O-bound concurrency — your server spends most of its time waiting for database queries and external API calls, not crunching numbers.

Async & Await Declarations

Coroutines

Declaring a function with async def creates a coroutine. Calling a coroutine does not run it — it returns a coroutine object. To actually execute it, you must await it:

import asyncio

async def fetch_data():
    print("Start fetching data...")
    await asyncio.sleep(2)   # non-blocking 2-second pause
    print("Data fetched!")
    return {"data": 123}

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())
# Start fetching data...
# (2 second pause)
# Data fetched!
# {'data': 123}

asyncio.sleep() is the async equivalent of time.sleep(). Unlike time.sleep(), it releases control back to the event loop during the wait, allowing other tasks to run.

The Event Loop and Task Scheduling

The event loop is the engine that drives async programs:

1. Run a task

   The event loop starts executing a coroutine.

2. Hit an await

   The coroutine reaches an await expression (e.g., a database query or network call).

3. Pause and switch

   The loop pauses the current coroutine and picks up another ready task.

4. Resume

   When the I/O operation completes, the loop resumes the original coroutine from where it paused.

Running Tasks Concurrently with asyncio.gather()

Without gathering, tasks run sequentially — the total time is the sum of all delays. With asyncio.gather(), tasks run concurrently — the total time is only as long as the slowest task:

import asyncio
import time

async def call_api(service_name: str, delay: int):
    print(f"Calling {service_name}...")
    await asyncio.sleep(delay)
    print(f"{service_name} done!")
    return f"{service_name} response"

async def main():
    start = time.time()

    results = await asyncio.gather(
        call_api("Auth Service",     2),
        call_api("Product Catalog",  1),
        call_api("Payment Gateway",  3),
    )

    elapsed = time.time() - start
    print(f"Results: {results}")
    print(f"Total time: {elapsed:.2f}s")   # ~3s instead of 6s

asyncio.run(main())

Without gather, these three calls would take 2 + 1 + 3 = 6 seconds. With gather, they run concurrently and finish in just 3 seconds — the duration of the slowest call.

create_task() for More Control

When you need to start a coroutine and continue working before its result is ready, use asyncio.create_task():

async def main():
    task = asyncio.create_task(call_api("Background Service", 2))

    # Do other work while the task runs in the background
    print("Doing something else...")
    await asyncio.sleep(1)

    # Now wait for the task to finish
    result = await task
    print(result)

Why FastAPI Uses Async

FastAPI is built on ASGI (Asynchronous Server Gateway Interface) and natively supports async def route handlers:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Simulates an async database query
    await asyncio.sleep(0.1)
    return {"user_id": user_id, "name": "Alice"}

When a client request hits this endpoint:

1. FastAPI calls the coroutine.
2. The coroutine hits await asyncio.sleep(0.1) (or a real database query).
3. The event loop pauses this coroutine and immediately starts handling the next incoming request.
4. When the database responds, the original coroutine resumes and sends its HTTP response.

A single process with a single thread can serve thousands of concurrent requests this way — without the overhead of threads or processes.

Use async def for route handlers that perform I/O — database queries, HTTP calls, file reads. Use regular def for pure CPU computation. FastAPI handles both correctly, but mixing them incorrectly (e.g., calling a blocking time.sleep() inside an async def handler) will block the event loop and hurt performance.

Never call blocking (synchronous) I/O operations directly inside an async def function. Replace time.sleep() with await asyncio.sleep(), and use async database drivers (like asyncpg or SQLAlchemy async) instead of their synchronous counterparts.