Python Asyncio Deep Dive: Coroutines, Tasks, and Event Loops in 2026

+---------------------------------------------------------------------------------+
|                                 Single OS Thread                                |
|                                                                                 |
|   +-------------------------------------------------------------------------+   |
|   |                           Asyncio Event Loop                            |   |
|   |                                                                         |   |
|   |   1. Poll OS Kernel Socket Ready State (epoll / kqueue)                 |   |
|   |   2. Resume Waiting Coroutines (send next() value)                      |   |
|   |   3. Suspend at 'await' boundary (yield socket descriptor to loop)      |   |
|   |   4. Run Scheduled Timers and Callbacks                                 |   |
|   +-------------------------------------------------------------------------+   |
|                                        |                                        |
|                                        v                                        |
|   +-------------------------------------------------------------------------+   |
|   |                          OS Kernel Non-Blocking I/O                     |   |
|   |   [ Socket 1: Reading ]  [ Socket 2: Writing ]  [ Socket 3: Connected ] |   |
|   +-------------------------------------------------------------------------+   |
+---------------------------------------------------------------------------------+

import asyncio

async def fetch_user(user_id: int):
    await asyncio.sleep(0.5)
    if user_id == 2:
        raise ValueError("User database connection timeout!")
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_orders(user_id: int):
    # This keeps running and executing even after fetch_user fails!
    await asyncio.sleep(2.0)
    print("Orders fetched (wasted database compute!)")
    return ["Order_101", "Order_102"]

async def main_legacy():
    try:
        results = await asyncio.gather(
            fetch_user(2),
            fetch_orders(2)
        )
    except ValueError as e:
        print(f"Caught error: {e}")
        # fetch_orders is still running in the background!

import asyncio

async def fetch_user(user_id: int):
    await asyncio.sleep(0.5)
    if user_id == 2:
        raise ValueError("User database connection failed!")
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_orders(user_id: int):
    try:
        await asyncio.sleep(2.0)
        return ["Order_101", "Order_102"]
    except asyncio.CancelledError:
        print("fetch_orders was cleanly cancelled because fetch_user failed!")
        raise

async def main_modern():
    try:
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(fetch_user(2))
            task2 = tg.create_task(fetch_orders(2))
        
        # Execution reaches here only if both tasks succeed
        print(task1.result(), task2.result())
    except* ValueError as eg:
        # Python 3.11+ exception group pattern matching
        for error in eg.exceptions:
            print(f"Handled error in TaskGroup: {error}")

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

import asyncio
import time

# WRONG: Synchronous blocking call inside async function!
async def bad_endpoint():
    # Freezes the ENTIRE event loop for 2 full seconds!
    # No other requests can be accepted or processed during this window.
    time.sleep(2.0)
    return {"status": "ok"}

import asyncio
from PIL import Image

def resize_image_sync(filepath: str, output_path: str):
    # CPU-heavy image compression
    with Image.open(filepath) as img:
        img.thumbnail((800, 800))
        img.save(output_path, "JPEG", quality=85)

async def handle_image_upload(filepath: str, output_path: str):
    # Cleanly offloaded to background thread pool
    await asyncio.to_thread(resize_image_sync, filepath, output_path)
    return {"status": "optimized"}

import asyncio
import aiohttp

class BoundedAPIScraper:
    def __init__(self, max_concurrent_requests: int = 25):
        self.semaphore = asyncio.Semaphore(max_concurrent_requests)
        self.session: aiohttp.ClientSession | None = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def fetch_item(self, item_id: int) -> dict:
        async with self.semaphore:
            # At most 25 requests will enter this block concurrently
            url = f"https://api.example.com/v1/items/{item_id}"
            assert self.session is not None
            async with self.session.get(url) as response:
                return await response.json()

    async def process_all(self, item_ids: list[int]) -> list[dict]:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(self.fetch_item(i)) for i in item_ids]
        return [t.result() for t in tasks]

uv add uvloop

import asyncio
import sys

# Configure uvloop as the default event loop policy on Unix
if sys.platform != "win32":
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def main():
    print("Running on ultra-fast libuv event loop!")

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