Async Python Skill

Master asynchronous Python programming for high-performance concurrent applications.

Asyncio Fundamentals

Basic Async/Await

import asyncio

async def fetch_data(url: str) -> dict:
    """Simulated async fetch."""
    await asyncio.sleep(1)  # Simulates I/O
    return {"url": url, "data": "..."}

async def main():
    # Sequential (slow)
    result1 = await fetch_data("https://api1.com")
    result2 = await fetch_data("https://api2.com")

    # Concurrent (fast)
    results = await asyncio.gather(
        fetch_data("https://api1.com"),
        fetch_data("https://api2.com"),
    )
    return results

# Run the event loop
asyncio.run(main())

Task Groups (Python 3.11+)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("https://api1.com"))
        task2 = tg.create_task(fetch_data("https://api2.com"))

    # All tasks complete when exiting context
    return task1.result(), task2.result()

HTTP Requests with aiohttp

Client Session

import aiohttp
import asyncio

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

async def main():
    # Reuse session for connection pooling
    async with aiohttp.ClientSession() as session:
        urls = [
            "https://api.example.com/users",
            "https://api.example.com/posts",
            "https://api.example.com/comments",
        ]

        tasks = [fetch_json(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"Error fetching {url}: {result}")
            else:
                print(f"Got {len(result)} items from {url}")

asyncio.run(main())

With Retry Logic

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def fetch_with_retry(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
        response.raise_for_status()
        return await response.json()

Async Database Access

With asyncpg (PostgreSQL)

import asyncpg

async def main():
    # Connection pool
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db",
        min_size=5,
        max_size=20,
    )

    async with pool.acquire() as conn:
        # Single query
        user = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )

        # Multiple rows
        users = await conn.fetch(
            "SELECT * FROM users WHERE active = $1", True
        )

        # Transaction
        async with conn.transaction():
            await conn.execute(
                "UPDATE users SET balance = balance - $1 WHERE id = $2",
                100, sender_id
            )
            await conn.execute(
                "UPDATE users SET balance = balance + $1 WHERE id = $2",
                100, receiver_id
            )

    await pool.close()

With SQLAlchemy Async

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User).where(User.active == True))
        return result.scalars().all()

Concurrency Patterns

Semaphore for Rate Limiting

async def fetch_with_limit(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url: str):
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.text()

    return await asyncio.gather(*[fetch_one(url) for url in urls])

Producer-Consumer Queue

async def producer(queue: asyncio.Queue, items: list):
    for item in items:
        await queue.put(item)
    # Signal completion
    await queue.put(None)

async def consumer(queue: asyncio.Queue, worker_id: int):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        # Process item
        await process(item)
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=100)

    # Start consumers
    consumers = [
        asyncio.create_task(consumer(queue, i))
        for i in range(5)
    ]

    # Start producer
    await producer(queue, items)

    # Wait for queue to be processed
    await queue.join()

    # Cancel consumers
    for c in consumers:
        c.cancel()

Timeout Handling

async def with_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=5.0
        )
    except asyncio.TimeoutError:
        print("Operation timed out")
        result = None
    return result

FastAPI Integration

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.pool = await asyncpg.create_pool(DATABASE_URL)
    yield
    # Shutdown
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

async def get_db():
    async with app.state.pool.acquire() as conn:
        yield conn

@app.get("/users/{user_id}")
async def get_user(user_id: int, db = Depends(get_db)):
    user = await db.fetchrow(
        "SELECT * FROM users WHERE id = $1", user_id
    )
    return user

Performance Tips

# 1. Use uvloop for faster event loop
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 2. Batch database operations
async def insert_many(items: list[dict]):
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO items (name, value) VALUES ($1, $2)",
            [(i["name"], i["value"]) for i in items]
        )

# 3. Stream large responses
async def stream_response():
    async with session.get(url) as response:
        async for chunk in response.content.iter_chunked(8192):
            yield chunk

Anti-Patterns

❌ Blocking I/O in async code (

time.sleep

, sync HTTP) ❌ Creating new sessions per request ❌ Not closing resources properly ❌ Ignoring exceptions in gather ❌ Running CPU-bound work in event loop

✅ Use async versions (

asyncio.sleep

,

aiohttp

) ✅ Reuse connection pools ✅ Use context managers ✅ Handle exceptions:

return_exceptions=True

✅ Use

run_in_executor

for CPU work

# CPU-bound work
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
    None,  # Default executor
    cpu_intensive_function,
    arg1, arg2
)