⚡ Asyncio & Async Programming

What is Asynchronous? Why Do We Need It?

Suppose you need to request 100 web pages simultaneously. The synchronous approach does them one by one — wait for the first page to return, then request the second. If each request takes 1 second, it takes 100 seconds in total.

The asynchronous approach is send a request without waiting for a reply, then immediately send the next one — 100 requests are sent almost simultaneously, wait for all replies to come back together, taking only 1-2 seconds total.

💡 Tip: 💡 Analogy: Synchronous = You order one dish at a restaurant, stand at the kitchen door waiting for it to be ready before ordering the next one. Asynchronous = You order all 100 dishes at once, then sit down and wait for the kitchen to bring them all together.

asyncio Basics

import asyncio

# Define an async function (coroutine)
async def fetch_data(url, delay):
    print(f"Start request {url}")
    await asyncio.sleep(delay)  # Simulate network request (non-blocking wait)
    print(f"Finish request {url}")
    return f"{url} data"

# Run coroutines
async def main():
    # ❌ Synchronous way: one by one, total time = sum of all delays
    # result1 = await fetch_data("api/users", 2)
    # result2 = await fetch_data("api/orders", 3)
    # Total time: 5 seconds

    # ✅ Asynchronous way: concurrent execution, total time = the longest one
    results = await asyncio.gather(
        fetch_data("api/users", 2),
        fetch_data("api/orders", 3),
        fetch_data("api/products", 1),
    )
    # Total time: 3 seconds (the longest one)

    for r in results:
        print(r)

asyncio.run(main())

Start request api/users
Start request api/orders
Start request api/products
Finish request api/products     ← after 1 second
Finish request api/users        ← after 2 seconds
Finish request api/orders       ← after 3 seconds (not 2+3=5 seconds!)

aiohttp Concurrent Web Scraper

Real-world scenario: concurrently scraping 100 web pages. Use aiohttp instead of requests:

import asyncio
import aiohttp

async def fetch(session, url):
    """Fetch a single web page"""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            text = await resp.text()
            return {"url": url, "status": resp.status, "length": len(text)}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/404",
        "https://httpbin.org/html",
    ] * 25  # Simulate 100 URLs

    # Limit concurrency (to avoid getting IP banned)
    semaphore = asyncio.Semaphore(20)

    async def limited_fetch(session, url):
        async with semaphore:
            return await fetch(session, url)

    async with aiohttp.ClientSession() as session:
        # Concurrently fetch all URLs (max 20 at a time)
        tasks = [limited_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    # Aggregate results
    success = sum(1 for r in results if r.get("status") == 200)
    failed = len(results) - success
    print(f"✅ Success: {success}, ❌ Failed: {failed}")

asyncio.run(main())

httpx — Modern Async HTTP Client

import asyncio
import httpx

async def main():
    # httpx supports both synchronous and asynchronous use
    async with httpx.AsyncClient() as client:
        # Concurrent requests
        tasks = [
            client.get("https://api.github.com/users/torvalds"),
            client.get("https://api.github.com/users/gvanrossum"),
            client.get("https://api.github.com/users/rosalindfranklin"),
        ]
        responses = await asyncio.gather(*tasks)

        for resp in responses:
            data = resp.json()
            print(f"{data['login']}: {data['public_repos']} repos")

asyncio.run(main())

💡 Tip: 💡 httpx vs aiohttp: httpx's API is more similar to requests (lower learning curve) and also supports HTTP/2. aiohttp is lighter and slightly more performant. httpx is recommended for new projects; no need to switch if you already have aiohttp code.

Async Task Management

import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return "slow done"

async def fast_task():
    await asyncio.sleep(1)
    return "fast done"

async def main():
    # 1. gather: wait for all tasks to complete
    results = await asyncio.gather(slow_task(), fast_task())

    # 2. wait_for: set a timeout (raises TimeoutError if exceeded)
    try:
        result = await asyncio.wait_for(slow_task(), timeout=3)
    except asyncio.TimeoutError:
        print("Task timed out!")

    # 3. as_completed: process in completion order
    tasks = [fast_task(), slow_task(), fast_task()]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Completed: {result}")

    # 4. create_task: create background tasks
    task = asyncio.create_task(slow_task())
    # Can do other things here...
    result = await task  # Await when you need the result

    # 5. Semaphore: limit concurrency
    sem = asyncio.Semaphore(5)  # Max 5 concurrent
    async def limited():
        async with sem:
            return await fast_task()

asyncio.run(main())

Async Generators and Context Managers

import asyncio

# Async generator (use async for to iterate)
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)  # Simulate async operation
        yield i

async def main():
    async for num in async_range(5):
        print(num)  # 0, 1, 2, 3, 4

    # Async context manager (use async with)
    class AsyncDB:
        async def __aenter__(self):
            print("Connecting to database...")
            await asyncio.sleep(0.5)
            return self

        async def __aexit__(self, *args):
            print("Closing database connection...")
            await asyncio.sleep(0.1)

        async def query(self, sql):
            await asyncio.sleep(0.2)
            return [{"id": 1, "name": "test"}]

    async with AsyncDB() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

asyncio.run(main())