Async Python

1 — The Mental Model: Same Concept, Different Runtime

You already know async/await from JavaScript. The concept is identical — a single-threaded event loop that cooperative-multitasks coroutines. The syntax is nearly the same too. The differences are in how you run it and what the ecosystem looks like.

// JS runs in an event loop automatically
// Any async function can be called anywhere

async function fetchUser(id) {
  const res = await fetch(`/users/${id}`);
  return res.json();
}

// Top-level await (in ESM)
const user = await fetchUser(1);

// Promise.all
const [u1, u2] = await Promise.all([
  fetchUser(1),
  fetchUser(2),
]);

import asyncio
import aiohttp  # async HTTP (pip install aiohttp)

async def fetch_user(session, id):
    async with session.get(f"/users/{id}") as res:
        return await res.json()

async def main():
    async with aiohttp.ClientSession() as session:
        user = await fetch_user(session, 1)

        # asyncio.gather = Promise.all
        u1, u2 = await asyncio.gather(
            fetch_user(session, 1),
            fetch_user(session, 2),
        )

# Must explicitly run the event loop
asyncio.run(main())

2 — Coroutines, Tasks, and Futures

import asyncio

# A coroutine function — calling it returns a coroutine object
async def say_hello(name, delay):
    await asyncio.sleep(delay)   # non-blocking sleep
    print(f"Hello, {name}!")

# A coroutine object — not running yet
coro = say_hello("Alice", 1)

# To run it, you need the event loop
asyncio.run(say_hello("Alice", 1))

# Inside an async context:
async def main():
    # await — run and wait for result
    await say_hello("Alice", 1)

    # Task — schedule to run concurrently (like Promise in JS)
    task = asyncio.create_task(say_hello("Bob", 0.5))
    # ... do other things ...
    await task   # wait for it to complete

    # gather — run multiple concurrently (Promise.all)
    await asyncio.gather(
        say_hello("Alice", 2),
        say_hello("Bob", 1),
        say_hello("Carol", 0.5),
    )
    # All three run concurrently — total time ≈ 2s, not 3.5s

3 — async with & async for

Python extends context managers and iterators to be async too.

import asyncio
import aiofiles   # pip install aiofiles

# async with — async context manager
async def read_file():
    async with aiofiles.open("data.txt") as f:
        content = await f.read()
    return content

# async for — async iterator
async def stream_data():
    async for chunk in some_async_stream():
        process(chunk)

# Async generator (yields from an async context)
async def paginate(url):
    page = 1
    async with aiohttp.ClientSession() as session:
        while True:
            async with session.get(url, params={"page": page}) as r:
                data = await r.json()
                if not data:
                    return
                for item in data:
                    yield item    # yield inside async def = async generator
            page += 1

async for user in paginate("https://api.example.com/users"):
    print(user)

4 — Error Handling in Async

import asyncio

# try/except works exactly the same
async def risky():
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get("https://api.example.com/data",
                                   timeout=aiohttp.ClientTimeout(total=5)) as r:
                r.raise_for_status()
                return await r.json()
        except aiohttp.ClientTimeout:
            print("Request timed out")
        except aiohttp.ClientError as e:
            print(f"HTTP error: {e}")

# asyncio.gather with error handling
async def main():
    results = await asyncio.gather(
        fetch(1),
        fetch(2),
        fetch(3),
        return_exceptions=True   # don't let one failure cancel others
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"One failed: {r}")
        else:
            print(r)

# Timeout (Python 3.11+)
async def with_timeout():
    async with asyncio.timeout(5.0):   # TimeoutError if >5s
        result = await slow_operation()

5 — sync vs async: The Critical Rule

import asyncio
import time

# BAD — blocks the entire event loop
async def bad():
    time.sleep(2)         # ← blocks! All other coroutines freeze
    data = open("file")   # ← blocking I/O
    result = heavy_computation()  # ← CPU-bound work

# GOOD — run blocking code in a thread pool
async def good():
    loop = asyncio.get_event_loop()

    # I/O bound blocking → run in thread pool
    data = await loop.run_in_executor(None, blocking_io_fn)

    # CPU bound → run in process pool
    from concurrent.futures import ProcessPoolExecutor
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_fn)

# Even simpler with asyncio.to_thread (Python 3.9+)
async def good_simple():
    result = await asyncio.to_thread(blocking_function, arg1, arg2)

6 — Complete Real-World Example

"""
Fetch GitHub user data for multiple users concurrently.
Run with: python3 script.py
"""
import asyncio
import aiohttp

USERS = ["torvalds", "gvanrossum", "antirez"]

async def fetch_user(session, username):
    url = f"https://api.github.com/users/{username}"
    async with session.get(url) as response:
        response.raise_for_status()
        data = await response.json()
        return {"name": data["name"], "repos": data["public_repos"]}

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, u) for u in USERS]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for user, result in zip(USERS, results):
        if isinstance(result, Exception):
            print(f"  {user}: FAILED — {result}")
        else:
            print(f"  {result['name']}: {result['repos']} repos")

if __name__ == "__main__":
    asyncio.run(main())