Mastering Python AsyncIO: A Complete Guide

import asyncio

async def fetch(url):
    # This yields control during I/O wait
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    urls = ["https://api.example.com/1", "https://api.example.com/2"]
    # Concurrent, not parallel
    results = await asyncio.gather(*[fetch(u) for u in urls])
    return results

asyncio.run(main())

# WRONG: sequential despite async def
async def slow():
    await asyncio.sleep(1)
    return "done"

results = await asyncio.gather(slow(), slow())  # 2 seconds

# RIGHT: concurrent via Task creation
async def main():
    t1 = asyncio.create_task(slow())
    t2 = asyncio.create_task(slow())
    return await asyncio.gather(t1, t2)  # 1 second

async def robust_fetch(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    # All succeeded or all cancelled
    return [t.result() for t in tasks]

# Automatic cleanup on any failure
# No try/finally boilerplate needed

# Per-operation timeout
async def safe_fetch(url):
    try:
        return await asyncio.wait_for(fetch(url), timeout=5.0)
    except asyncio.TimeoutError:
        logger.warning(f"Timeout: {url}")
        return None

# Scope timeout (Python 3.11+)
async def batch_with_deadline(urls):
    async with asyncio.timeout(30.0):
        return await robust_fetch(urls)

async def producer(queue, items):
    for item in items:
        await queue.put(item)  # Blocks if queue full
    await queue.put(None)  # Sentinel

async def consumer(queue, results):
    while True:
        item = await queue.get()
        if item is None:
            break
        results.append(await process(item))
        queue.task_done()

async def main():
    q = asyncio.Queue(maxsize=100)
    await asyncio.gather(
        producer(q, range(1000)),
        consumer(q, [])
    )