AtomLearnAtomLearn
DashboardGoalsPathAchievementsReviewSign In
Python CoreNot Started

async / await

Coroutines, the event loop, asyncio, and async I/O patterns

0%
Knowledge0%
Learn & Drill
Fluency0%
Drill & Speed
Retention0%
Mastery & Review
Confidence0%
All modes
Practice

Free tier: read the explanation here. Upgrade to Pro for Drills, Speed challenges & Mastery badges.

Upgrade
Knowledge
Fluency
Retention

Knowledge Debt detected

You can study this freely — but your score may plateau if these foundations have gaps. The Mastery badge requires them to be solid.

Explanation

Python's async/await lets you write concurrent I/O-bound code without threads. It's the foundation of FastAPI, aiohttp, and modern Python services.

async/await — tasks interleave at await points (single thread)TaskTime →fetch_a()CPUawait I/O…CPUfetch_b()CPUawait I/O…CPUfetch_c()CPUawait I/O…CPUSequential(no async):← much slowerawait suspends the current coroutine and lets the event loop run other tasks — CPU is never idle waiting for I/O

The core concept — :coroutine[A function defined with async def that can suspend execution using await]s:

python import asyncio async def fetch_user(user_id: int) -> dict: await asyncio.sleep(0.1) # simulate I/O without blocking return {"id": user_id, "name": "Alice"} # Must be run inside an event loop asyncio.run(fetch_user(1))

await suspends the current coroutine and yields control back to the . The event loop can then run other coroutines while waiting.

Running multiple coroutines :concurrent[Multiple tasks making progress by interleaving execution on one thread during I/O waits]ly:

python import asyncio async def main(): # Sequential — total ~0.3s a = await fetch(1) b = await fetch(2) c = await fetch(3) # Concurrent — total ~0.1s a, b, c = await asyncio.gather( fetch(1), fetch(2), fetch(3) )

Async context managers and iterators:

python async with aiohttp.ClientSession() as session: async with session.get(url) as response: data = await response.json() async for record in db.execute("SELECT * FROM users"): process(record)

Key rule: async is contagious — to use await, you must be inside an async def. Regular sync code cannot call coroutines directly.

When to use async vs threads:

  • Async: many concurrent I/O operations (HTTP calls, DB queries, file reads)
  • Threads: CPU-bound work or blocking third-party libraries that don't support async

Examples

Concurrent HTTP requests

asyncio.gather fires all requests simultaneously — much faster than sequential awaits

import asyncio
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
results = asyncio.run(fetch_all(urls))
print(f"Fetched {len(results)} posts concurrently")

Async context manager for DB connection

asyncpg is the async PostgreSQL driver used with FastAPI — same pattern as sync but non-blocking

import asyncio
import asyncpg

async def get_users():
    conn = await asyncpg.connect("postgresql://localhost/mydb")
    try:
        rows = await conn.fetch("SELECT id, name FROM users LIMIT 10")
        return [dict(r) for r in rows]
    finally:
        await conn.close()

users = asyncio.run(get_users())

How well did you understand this?

Next in Python Core

Concurrency Patterns

Continue