Python async/await & asyncio

Also Known As

TL;DR

Explanation

Python's asyncio provides a single-threaded event loop similar to Node.js. async def functions are coroutines — they don't run until awaited or scheduled. await suspends the coroutine until the awaited coroutine/task finishes. asyncio.gather(*coros) runs coroutines concurrently. asyncio.run(main()) starts the event loop. For CPU-bound work, asyncio doesn't help — use multiprocessing or ThreadPoolExecutor (loop.run_in_executor). Key difference from threading: no GIL contention, but no true parallelism either — only concurrency for I/O. Libraries: httpx (async HTTP), asyncpg (async PostgreSQL), aiofiles (async file I/O), FastAPI (async web framework). Comparable to PHP's Fibers + Revolt, but asyncio is a mature standard library with rich ecosystem support.

Diagram

flowchart TD
    subgraph Sync_Python
        S1[Task A runs] --> S2[Task A waits for IO<br/>thread blocked]
        S2 --> S3[Task A resumes]
        S3 --> S4[Task B runs]
    end
    subgraph Async_Python
        A1[Task A starts] --> A2[await IO - suspended]
        A2 --> B1[Task B runs during A wait]
        B1 --> A3[Task A resumes]
        A3 & B1 -->|interleaved| FAST[Both complete faster]
    end
    subgraph asyncio
        LOOP2[Event loop<br/>asyncio.run]
        GATHER[asyncio.gather<br/>run concurrently]
        LOOP2 --> GATHER
    end
style A2 fill:#d29922,color:#fff
style FAST fill:#238636,color:#fff
style GATHER fill:#1f6feb,color:#fff

Common Misconception

Why It Matters

Common Mistakes

• Using blocking I/O inside async functions — blocks the entire event loop.
• Not awaiting coroutines — calling an async function without await returns a coroutine object, not the result.
• Using asyncio.run() inside an already-running event loop — raises RuntimeError.
• CPU-bound work in async functions — use ProcessPoolExecutor, not asyncio, for CPU-heavy tasks.

Code Examples

# Blocking call inside async — freezes event loop:
async def fetch_data():
    time.sleep(2)  # Blocks! Use: await asyncio.sleep(2)
    data = requests.get(url)  # Blocks! Use: await aiohttp session
    return data

import asyncio
import aiohttp

# Async function — returns a coroutine
async def fetch_user(session: aiohttp.ClientSession, user_id: int) -> dict:
    async with session.get(f'https://api.example.com/users/{user_id}') as resp:
        return await resp.json()

# Fetch multiple users concurrently
async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, uid) for uid in user_ids]
        return await asyncio.gather(*tasks)  # run concurrently

# Run:
users = asyncio.run(fetch_all_users([1, 2, 3, 4, 5]))

# async with — async context manager
# async for  — async iteration (e.g. streaming responses)
async def stream_lines(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            async for line in resp.content:
                yield line.decode('utf-8').strip()

Tags

🧠 Logic Engine

References

• https://docs.python.org/3/library/asyncio.html