Async/Await and Concurrency in Python with asyncio

If your program spends most of its time waiting — for a database query to come back, a file to finish reading, or an API to respond — asynchronous programming lets you do useful work during that waiting time instead of blocking idle. This is the foundation that makes FastAPI capable of handling thousands of simultaneous requests on a single process.

Concurrency vs Parallelism

These two terms are frequently confused. Understanding the difference tells you which tool to pick.

Concurrency (I/O-bound)

Multiple tasks make progress by taking turns. While one task waits for a response, another runs. All tasks share a single CPU core. This is asyncio’s domain.Use cases: web requests, database queries, file I/O, WebSocket connections.

Parallelism (CPU-bound)

Multiple tasks run at the exact same moment on different CPU cores. Python uses multiprocessing for this because the GIL prevents true thread-level parallelism.Use cases: image processing, matrix math, video encoding, machine learning training.

A web server like FastAPI lives in the concurrency world: each request mainly waits on the database. Async lets it serve request B while waiting for request A’s database query to return.

`async def` and `await`

Declaring a function with async def creates a coroutine. Calling it returns a coroutine object — nothing runs yet. To actually execute it and get the result, you await it inside another coroutine.

import asyncio

async def fetch_user(user_id: int) -> dict:
    print(f"Fetching user {user_id}...")
    await asyncio.sleep(1)           # simulates a 1-second database query
    print(f"User {user_id} ready")
    return {"id": user_id, "name": f"User {user_id}"}

async def main():
    user = await fetch_user(42)
    print(user)

asyncio.run(main())   # starts the event loop and runs main()
# Fetching user 42...
# User 42 ready
# {'id': 42, 'name': 'User 42'}

You can only await inside an async def function. Regular (synchronous) functions cannot use await. If you call asyncio.run() from within a running event loop (e.g., inside a Jupyter notebook), use await main() directly instead.

The Event Loop

The event loop is the scheduler that drives async code. It maintains a queue of coroutines, runs each one until it hits an await, pauses it, and switches to the next ready task. When the I/O operation completes, the original task is resumed.

Event Loop
   │
   ├─ Run task A → hits await → pause A
   ├─ Run task B → hits await → pause B
   ├─ I/O for A completes → resume A → A finishes
   └─ I/O for B completes → resume B → B finishes

You start the event loop once at the top level with asyncio.run(). Inside async code, you never call asyncio.run() again — you just await.

Running Tasks Concurrently with `asyncio.gather()`

The real power of async appears when you fire off multiple tasks at the same time. asyncio.gather() takes several coroutines and runs them concurrently, returning all their results once every one has completed.

import asyncio
import time

async def call_service(name: str, delay: float) -> str:
    print(f"→ Calling {name}")
    await asyncio.sleep(delay)
    print(f"← {name} responded")
    return f"{name}: OK"

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

    # Run all three calls concurrently — total time ≈ max(2, 1, 3) = 3 seconds
    results = await asyncio.gather(
        call_service("Auth Service",    delay=2),
        call_service("Product Catalog", delay=1),
        call_service("Payment Gateway", delay=3),
    )

    elapsed = time.perf_counter() - start
    print(f"\nResults: {results}")
    print(f"Total time: {elapsed:.2f}s")   # ≈ 3.00s, not 6.00s

asyncio.run(main())

Without gather, each await would run sequentially (6 seconds total). With gather, all three run concurrently (3 seconds total — limited only by the slowest task).

`asyncio.create_task()` for Fire-and-Forget

When you need a task to start immediately without waiting for it right now, use create_task():

async def background_job(label: str):
    await asyncio.sleep(2)
    print(f"Background job '{label}' done")

async def main():
    # Start the background task without awaiting it yet
    task = asyncio.create_task(background_job("cleanup"))

    print("Doing other work...")
    await asyncio.sleep(0.5)
    print("Still doing work...")

    await task   # wait for the background task to finish before exiting

asyncio.run(main())
# Doing other work...
# Still doing work...
# Background job 'cleanup' done

`async for` and `async with`

Use async for to iterate over an async generator (one that yields items over time, like a streaming database cursor) and async with to use an async context manager.

import asyncio

# Async generator — yields one item per second
async def streaming_data(n: int):
    for i in range(n):
        await asyncio.sleep(0.5)
        yield f"record_{i}"

async def main():
    async for record in streaming_data(4):
        print(record)

asyncio.run(main())
# record_0
# record_1
# record_2
# record_3

# Async context manager — e.g., an async database connection
async with async_db.connect("postgres://...") as conn:
    rows = await conn.fetch("SELECT * FROM users")

Real-World Example: Fetch Multiple API Endpoints

import asyncio
import json

# In real code you'd use: import httpx
# Here we simulate network calls with asyncio.sleep

async def fetch_endpoint(session, url: str) -> dict:
    await asyncio.sleep(0.3)   # simulated network latency
    return {"url": url, "status": 200, "data": "..."}

async def fetch_all(urls: list[str]) -> list[dict]:
    # Use asyncio.gather to fire all requests concurrently
    tasks = [fetch_endpoint(None, url) for url in urls]
    return await asyncio.gather(*tasks)

async def main():
    endpoints = [
        "https://api.example.com/users",
        "https://api.example.com/products",
        "https://api.example.com/orders",
        "https://api.example.com/analytics",
    ]

    import time
    start = time.perf_counter()
    results = await fetch_all(endpoints)
    elapsed = time.perf_counter() - start

    for r in results:
        print(f"{r['url']} → {r['status']}")
    print(f"\nAll {len(results)} requests completed in {elapsed:.2f}s")
    # ≈ 0.30s regardless of how many endpoints — they all run in parallel

asyncio.run(main())

FastAPI and Async

FastAPI is built on Starlette, an ASGI framework, and runs on an async event loop. Every async def route handler you write is a coroutine that FastAPI awaits inside its event loop:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    # While this query waits, FastAPI handles other incoming requests
    product = await db.fetch_one("SELECT * FROM products WHERE id = ?", product_id)
    return product

You can mix sync and async route handlers in FastAPI. Use async def for handlers that call async libraries (async database drivers, httpx, etc.) and plain def for CPU-bound or purely synchronous operations. FastAPI runs plain def handlers in a thread pool automatically.

Common Pitfalls

Awaiting synchronous blocking calls

await only yields control for async operations. Calling a blocking synchronous function inside an async function (e.g., time.sleep(), synchronous requests.get()) blocks the entire event loop.

# ❌ Blocks the event loop — no other tasks can run during sleep
async def bad():
    time.sleep(2)

# ✅ Yields control — other tasks run while waiting
async def good():
    await asyncio.sleep(2)

Forgetting to await a coroutine

Calling an async function without await gives you a coroutine object, not the result. Python 3 will emit a RuntimeWarning.

result = fetch_user(42)    # ❌ coroutine object, not a dict
result = await fetch_user(42)  # ✅ actual dict

Core Python

Advanced Python

Pydantic

Async/Await and Concurrency in Python with asyncio

Concurrency vs Parallelism

Concurrency (I/O-bound)

Parallelism (CPU-bound)

`async def` and `await`

The Event Loop

Running Tasks Concurrently with `asyncio.gather()`

`asyncio.create_task()` for Fire-and-Forget

`async for` and `async with`

Real-World Example: Fetch Multiple API Endpoints

FastAPI and Async

Common Pitfalls

​Concurrency vs Parallelism

Concurrency (I/O-bound)

Parallelism (CPU-bound)

​async def and await

​The Event Loop

​Running Tasks Concurrently with asyncio.gather()

​asyncio.create_task() for Fire-and-Forget

​async for and async with

​Real-World Example: Fetch Multiple API Endpoints

​FastAPI and Async

​Common Pitfalls

Concurrency vs Parallelism

`async def` and `await`

The Event Loop

Running Tasks Concurrently with `asyncio.gather()`

`asyncio.create_task()` for Fire-and-Forget

`async for` and `async with`

Real-World Example: Fetch Multiple API Endpoints

FastAPI and Async

Common Pitfalls