Python Async and asyncio: Non-Blocking Code Complete Guide

Python Async and asyncio

Python's asyncio enables non-blocking I/O — running thousands of operations concurrently in a single thread. Essential for high-performance APIs and microservices.

import asyncio, aiohttp, time

async def fetch(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

async def main():
    # Sequential: 3 seconds (1s each)
    r1 = await fetch('https://api.example.com/users/1')
    r2 = await fetch('https://api.example.com/users/2')

    # Concurrent: ~1 second (all run simultaneously!)
    r1, r2, r3 = await asyncio.gather(
        fetch('https://api.example.com/users/1'),
        fetch('https://api.example.com/users/2'),
        fetch('https://api.example.com/users/3'),
    )

asyncio.run(main())

Tasks and Cancellation

async def worker(name: str, delay: float):
    print(f"{name} starting")
    await asyncio.sleep(delay)
    print(f"{name} done")

async def run():
    tasks = [
        asyncio.create_task(worker("A", 1.0)),
        asyncio.create_task(worker("B", 0.5)),
        asyncio.create_task(worker("C", 2.0)),
    ]
    await asyncio.gather(*tasks)
    # B finishes first, then A, then C

    # Cancel a task
    task = asyncio.create_task(worker("D", 100.0))
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("D was cancelled")