Python Async/Await in Practice: asyncio, FastAPI, and Common Pitfalls

import asyncio


async def fetch_user(user_id: int) -> dict:
    # Simulate a database query.
    await asyncio.sleep(0.05)
    return {"id": user_id, "name": "Alice"}


async def main() -> None:
    user = await fetch_user(1)
    print(user)


asyncio.run(main())

import asyncio
import time


async def fetch_from_service(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} responded after {delay}s"


async def main() -> None:
    start = time.monotonic()

    # Sequential: total time = 0.3 + 0.2 + 0.4 = 0.9s
    # r1 = await fetch_from_service("users", 0.3)
    # r2 = await fetch_from_service("orders", 0.2)
    # r3 = await fetch_from_service("inventory", 0.4)

    # Concurrent with gather: total time = max(0.3, 0.2, 0.4) = ~0.4s
    r1, r2, r3 = await asyncio.gather(
        fetch_from_service("users", 0.3),
        fetch_from_service("orders", 0.2),
        fetch_from_service("inventory", 0.4),
    )

    elapsed = time.monotonic() - start
    print(f"All done in {elapsed:.2f}s")
    print(r1, r2, r3)


asyncio.run(main())

import asyncio


async def background_job() -> None:
    await asyncio.sleep(1)
    print("background job done")


async def main() -> None:
    # Fire-and-forget: start the job but don't wait for it.
    task = asyncio.create_task(background_job())

    # Do other work while background_job runs.
    await asyncio.sleep(0.1)
    print("main work done")

    # Optionally wait for it before shutdown.
    await task


asyncio.run(main())

import asyncio
import time
import requests  # synchronous HTTP library


async def bad_handler() -> str:
    # This blocks the ENTIRE event loop for 2 seconds.
    # No other coroutine can run while this sleeps.
    time.sleep(2)
    response = requests.get("https://api.example.com/data")  # also blocking
    return response.text


async def good_handler() -> str:
    # Non-blocking sleep: yields control to the event loop.
    await asyncio.sleep(2)
    # Non-blocking HTTP: use an async library.
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as response:
            return await response.text()

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)


def blocking_cpu_work(n: int) -> int:
    # Simulate CPU-bound work.
    time.sleep(0.5)
    return n * n


async def main() -> None:
    loop = asyncio.get_running_loop()
    # Run blocking_cpu_work in a thread without freezing the event loop.
    result = await loop.run_in_executor(executor, blocking_cpu_work, 42)
    print(result)

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator


@asynccontextmanager
async def managed_connection(host: str) -> AsyncGenerator[dict, None]:
    print(f"connecting to {host}")
    await asyncio.sleep(0.01)  # simulate async connect
    conn = {"host": host, "open": True}
    try:
        yield conn
    finally:
        conn["open"] = False
        print(f"closed connection to {host}")


async def streamed_results(query: str) -> AsyncGenerator[dict, None]:
    # Async generator: yields rows as they arrive from the database.
    for i in range(5):
        await asyncio.sleep(0.01)
        yield {"row": i, "query": query}


async def main() -> None:
    async with managed_connection("db.internal") as conn:
        async for row in streamed_results("SELECT * FROM users"):
            print(row)


asyncio.run(main())

import asyncio
import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

SERVICES = {
    "user": "https://internal.example.com/users",
    "order": "https://internal.example.com/orders",
    "inventory": "https://internal.example.com/inventory",
}


async def fetch_json(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url, timeout=2.0)
    response.raise_for_status()
    return response.json()


@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        try:
            user, orders, inventory = await asyncio.gather(
                fetch_json(client, f"{SERVICES['user']}/{user_id}"),
                fetch_json(client, f"{SERVICES['order']}?user_id={user_id}"),
                fetch_json(client, f"{SERVICES['inventory']}?user_id={user_id}"),
                return_exceptions=True,
            )
        except Exception as exc:
            raise HTTPException(status_code=502, detail=str(exc)) from exc

    return {"user": user, "orders": orders, "inventory": inventory}