0xdarkmatter/claude-mods/python-async-ops

import asyncio
# Coroutine (must be awaited)
async def fetch(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
# Entry point
async def main():
    result = await fetch("https://example.com")
    return result
asyncio.run(main())

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

async def fetch_with_limit(urls: list[str], limit: int = 10):
    """Limit concurrent requests."""
    semaphore = asyncio.Semaphore(limit)
    async def bounded_fetch(url):
        async with semaphore:
            return await fetch_one(url)
    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

async def process_items(items):
    """Structured concurrency with automatic cleanup."""
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item))
    # All tasks complete here, or exception raised

async def with_timeout():
    try:
        async with asyncio.timeout(5.0):  # Python 3.11+
            result = await slow_operation()
    except asyncio.TimeoutError:
        result = None
    return result

# WRONG - blocks event loop
async def bad():
    time.sleep(5)         # Never use time.sleep!
    requests.get(url)     # Blocking I/O!
# CORRECT
async def good():
    await asyncio.sleep(5)
    async with aiohttp.ClientSession() as s:
        await s.get(url)

# WRONG - orphaned task
async def bad():
    asyncio.create_task(work())  # May be garbage collected!
# CORRECT - keep reference
async def good():
    task = asyncio.create_task(work())
    await task

from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_connection():
    conn = await create_connection()
    try:
        yield conn
    finally:
        await conn.close()