Python Async Patterns: Deep Dive into Coroutines and Event Loops

1. Async Programming Basics

Python’s asyncio library provides an event-loop-based coroutine concurrency model, ideal for I/O-bound and high-concurrency applications. Understanding coroutines, the event loop, and task scheduling is key to mastering asynchronous programming.

2. Core Example

The following example demonstrates writing concurrent API requests using async/await, and executing multiple coroutines in parallel with asyncio.gather.

Python Async Example

python
import asyncio
import aiohttp
from typing import List

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    """Fetch URL content asynchronously"""
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: List[str]) -> List[dict]:
    """Fetch all URLs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/posts",
        "https://api.example.com/comments",
    ]
    results = await fetch_all(urls)
    for i, result in enumerate(results):
        status = "Success" if not isinstance(result, Exception) else "Failed"
        print(f"Request {i}: {status}")

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

Run Script

bash
#!/bin/bash
# Async program runner
python3 -m pip install aiohttp  # Install dependencies
python3 async_fetch.py          # Run async program

3. Best Practices

Use asyncio.create_task for background tasks, asyncio.Semaphore to control concurrency, and asyncio.timeout for timeout management. Avoid CPU-bound operations inside coroutines; use concurrent.futures for parallel computation instead.