hivemind: Skill for Claude Code

Async Python Patterns

Core Concepts

import asyncio

# Basic async function
async def fetch_user(user_id: int) -> User:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"/users/{user_id}") as resp:
            return await resp.json()

# Run from sync context
user = asyncio.run(fetch_user(123))

Concurrency Patterns

Run tasks in parallel (gather)

# All run at the same time — total time = slowest task
users, posts, comments = await asyncio.gather(
    fetch_user(user_id),
    fetch_posts(user_id),
    fetch_comments(user_id),
)

# With error handling — return exceptions instead of raising
results = await asyncio.gather(
    fetch_user(user_id),
    fetch_posts(user_id),
    return_exceptions=True,
)
for result in results:
    if isinstance(result, Exception):
        logger.error(f"Task failed: {result}")

Limit concurrency (Semaphore)

# Process 100 items but max 10 at a time
sem = asyncio.Semaphore(10)

async def process_with_limit(item):
    async with sem:
        return await process(item)

results = await asyncio.gather(*[process_with_limit(i) for i in items])

Timeout

try:
    result = await asyncio.wait_for(fetch_data(), timeout=10.0)
except asyncio.TimeoutError:
    logger.warning("Fetch timed out after 10s")
    result = None

Background tasks (fire and forget)

async def main():
    # Don't await — runs in background
    task = asyncio.create_task(send_notification(user_id))

    # But keep reference so GC doesn't kill it
    background_tasks = set()
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

Async context manager

class DatabasePool:
    async def __aenter__(self):
        self.pool = await asyncpg.create_pool(dsn)
        return self

    async def __aexit__(self, *args):
        await self.pool.close()

async with DatabasePool() as db:
    await db.pool.fetchrow("SELECT * FROM users WHERE id = $1", 1)