async-python-patterns

Q: How do I install async-python-patterns?

Run npx skills add https://github.com/nickcrew/claude-cortex --skill async-python-patterns in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/async-python-patterns/ (personal) or .claude/skills/async-python-patterns/ (project).

by nickcrewCode Quality & Workflow

Python asyncio and concurrent programming patterns for high-performance applications. Use when building async APIs, concurrent systems, or I/O-bound applications requiring non-blocking operations.

Installs: 84
Last 24h: —
Trending rank: —
Security: Not audited

Install

Works with every agent the skills CLI supports.

npx skills add https://github.com/nickcrew/claude-cortex --skill async-python-patterns

01
Install
Run the command in your project. The CLI asks which agents to install for.
02
Describe the task
Ask your agent for something this skill covers. It loads async-python-patterns automatically when relevant.
03
Review the result
The agent follows the SKILL.md steps, scripts and references bundled with the skill.

SKILL.md

The instructions your agent reads when the skill is triggered.

Async Python Patterns

Expert guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems.

When to Use This Skill

Building async web APIs (FastAPI, aiohttp, Sanic)
Implementing concurrent I/O operations (database, file, network)
Creating web scrapers with concurrent requests
Developing real-time applications (WebSocket servers, chat systems)
Processing multiple independent tasks simultaneously
Optimizing I/O-bound workloads requiring parallelism
Implementing async background tasks and task queues

Core Patterns

1. Basic Async/Await

Foundation for all async operations:

import asyncio

async def fetch_data(url: str) -> dict:
    """Fetch data asynchronously."""
    await asyncio.sleep(1)  # Simulate I/O
    return {"url": url, "data": "result"}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

Key concepts:

async def defines coroutines (pausable functions)
await yields control back to event loop
asyncio.run() is the entry point (Python 3.7+)
Single-threaded cooperative multitasking

2. Concurrent Execution with gather()

Execute multiple operations simultaneously:

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

# Speed: Sequential = 5s, Concurrent = 0.5s for 10 users

When to use:

Independent operations that can run in parallel
I/O-bound tasks (API calls, database queries)
Need all results before proceeding
Use return_exceptions=True to handle partial failures

3. Task Creation and Management

Background tasks that run independently:

import asyncio

async def background_task(name: str, delay: int):
    print(f"{name} started")
    await asyncio.sleep(delay)
    return f"Result from {name}"

async def main():
    # Create tasks (starts execution immediately)
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

    # Do other work while tasks run
    print("Doing other work")
    await asyncio.sleep(0.5)

    # Wait for results when needed
    result1, result2 = await task1, await task2

Differences:

await coroutine() - Waits immediately
asyncio.create_task() - Starts background execution
Tasks can be cancelled with task.cancel()

4. Error Handling

Robust error handling for concurrent operations:

import asyncio
from typing import List, Optional

async def safe_operation(item_id: int) -> Optional[dict]:
    try:
        await asyncio.sleep(0.1)
        if item_id % 3 == 0:
            raise ValueError(f"Item {item_id} failed")
        return {"id": item_id, "status": "success"}
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids: List[int]):
    # gather with return_exceptions=True continues on errors
    results = await asyncio.gather(
        *[safe_operation(iid) for iid in item_ids],
        return_exceptions=True
    )

    successful = [r for r in results if r and not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]

    print(f"Success: {len(successful)}, Failed: {len(failed)}")
    return successful

5. Timeout Handling

Prevent operations from hanging indefinitely:

import asyncio

async def with_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(5),
            timeout=2.0
        )
        print(result)
    except asyncio.TimeoutError:
        print("Operation timed out")
        # Handle timeout (retry, fallback, etc.)

Advanced Patterns

6. Async Context Managers

Proper resource management with async operations:

import asyncio

class AsyncDatabaseConnection:
    """Async database connection with automatic cleanup."""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.connection = None

    async def __aenter__(self):
        print("Opening connection")
        await asyncio.sleep(0.1)  # Simulate connection
        self.connection = {"dsn": self.dsn, "connected": True}
        return self.connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection")
        await asyncio.sleep(0.1)  # Simulate cleanup
        self.connection = None

async def query_database():
    async with AsyncDatabaseConnection("postgresql://localhost") as conn:
        # Connection automatically closed on exit
        return await perform_query(conn)

Use cases:

Database connections (asyncpg, motor)
HTTP sessions (aiohttp.ClientSession)
File I/O (aiofiles)
Locks and semaphores

7. Async Iterators and Generators

Stream data asynchronously:

import asyncio
from typing import AsyncIterator

async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]:
    """Fetch paginated data lazily."""
    for page in range(1, max_pages + 1):
        await asyncio.sleep(0.2)  # API call
        yield {
            "page": page,
            "url": f"{url}?page={page}",
            "data": [f"item_{page}_{i}" for i in range(5)]
        }

async def process_stream():
    async for page_data in fetch_pages("https://api.example.com", 10):
        # Process each page as it arrives (memory efficient)
        print(f"Processing page {page_data['page']}")

Benefits:

Memory efficient for large datasets
Start processing before all data arrives
Natural backpressure handling

8. Producer-Consumer with Queues

Coordinate work between producers and consumers:

import asyncio
from asyncio import Queue

async def producer(queue: Queue, producer_id: int, num_items: int):
    for i in range(num_items):
        item = f"Item-{producer_id}-{i}"
        await queue.put(item)
        await asyncio.sleep(0.1)
    await queue.put(None)  # Signal completion

async def consumer(queue: Queue, consumer_id: int):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break

        print(f"Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(0.2)
        queue.task_done()

async def run_pipeline():
    queue = Queue(maxsize=10)

    # 2 producers, 3 consumers
    producers = [asyncio.create_task(producer(queue, i, 5)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]

    await asyncio.gather(*producers)
    await queue.join()  # Wait for all items processed

    for c in consumers:
        c.cancel()

9. Rate Limiting with Semaphores

Control concurrent operations:

import asyncio
from typing import List

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:  # Only N operations at once
        print(f"Calling {url}")
        await asyncio.sleep(0.5)
        return {"url": url, "status": 200}

async def rate_limited_requests(urls: List[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [api_call(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

# Limits to 5 concurrent requests regardless of total URLs

Use cases:

API rate limiting (respect API quotas)
Database connection limits
File descriptor limits
Memory-constrained operations

10. Async Locks for Shared State

Thread-safe operations in async context:

import asyncio

class AsyncCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:
            current = self.value
            await asyncio.sleep(0.01)  # Simulate work
            self.value = current + 1

    async def get_value(self) -> int:
        async with self.lock:
            return self.value

Synchronization primitives:

Lock: Mutual exclusion
Event: Signal between tasks
Condition: Wait for condition
Semaphore: Limit concurrent access

Performance Best Practices

1. Use Connection Pools

Reuse connections for efficiency:

import aiohttp

async def with_connection_pool():
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [session.get(f"https://api.example.com/item/{i}")
                 for i in range(50)]
        return await asyncio.gather(*tasks)

2. Avoid Blocking the Event Loop

Run CPU-intensive work in executor:

import asyncio
import concurrent.futures

def blocking_operation(data):
    """CPU-intensive blocking operation."""
    import time
    time.sleep(1)
    return data * 2

async def run_in_executor(data):
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_operation, data)
        return result

Common blockers to avoid:

time.sleep() - Use asyncio.sleep()
Synchronous file I/O - Use aiofiles
Synchronous HTTP - Use aiohttp or httpx
Heavy computation - Use loop.run_in_executor()

3. Batch Operations

Process in chunks to control memory:

async def batch_process(items: List[str], batch_size: int = 10):
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        results = await asyncio.gather(*[process_item(item) for item in batch])
        print(f"Processed batch {i // batch_size + 1}")

Common Pitfalls

1. Forgetting await

# Wrong - returns coroutine, doesn't execute
result = async_function()

# Correct
result = await async_function()

2. Blocking the Event Loop

# Wrong - blocks entire event loop
import time
async def bad():
    time.sleep(1)

# Correct
async def good():
    await asyncio.sleep(1)

3. Not Handling Cancellation

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Cleanup resources
        raise  # Re-raise to propagate

4. Mixing Sync and Async

# Wrong
def sync_function():
    result = await async_function()  # SyntaxError

# Correct
def sync_function():
    result = asyncio.run(async_function())

Testing Async Code

Use pytest-asyncio for testing:

import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await fetch_data("https://api.example.com")
    assert result is not None

@pytest.mark.asyncio
async def test_with_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(5), timeout=1.0)

Resources

Python asyncio docs: https://docs.python.org/3/library/asyncio.html
aiohttp: Async HTTP client/server framework
FastAPI: Modern async web framework with automatic OpenAPI
asyncpg: High-performance async PostgreSQL driver
motor: Async MongoDB driver for Python
pytest-asyncio: Testing framework for async code

Best Practices Summary

Use asyncio.run() for entry point (Python 3.7+)
Always await coroutines to execute them
Use gather() for concurrent execution of independent tasks
Implement proper error handling with try/except and return_exceptions=True
Use timeouts to prevent hanging operations
Pool connections for better performance and resource management
Avoid blocking operations in async code (use executors if needed)
Use semaphores for rate limiting and resource control
Handle task cancellation properly with CancelledError
Test async code thoroughly with pytest-asyncio

FAQ

What does the async-python-patterns skill do?

Python asyncio and concurrent programming patterns for high-performance applications. Use when building async APIs, concurrent systems, or I/O-bound applications requiring non-blocking operations.

How do I install async-python-patterns?

Run npx skills add https://github.com/nickcrew/claude-cortex --skill async-python-patterns in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/async-python-patterns/ (personal) or .claude/skills/async-python-patterns/ (project).

Which AI agents can use async-python-patterns?

Any agent that supports the open Agent Skills format (a folder with a SKILL.md file), including Claude Code, Codex, Cursor, GitHub Copilot, Gemini CLI, Windsurf, Cline, OpenCode and more.

Is async-python-patterns safe to install?

No partner security audit is available yet. Skills can run commands and scripts, so read the SKILL.md and any bundled files before installing.

Who maintains async-python-patterns?

It is published by nickcrew in nickcrew/claude-cortex. It has 84 recorded installs.

More from nickcrew/claude-cortex

View all →

react-performance-optimization

nickcrew/claude-cortex

React performance optimization patterns using memoization, code splitting, and efficient rendering strategies. Use when optimizing slow React applications, reducing bundle size, or improving user experience with large datasets.

Frontend & Design1.9K installs

owasp-top-10

nickcrew/claude-cortex

OWASP Top 10 security vulnerabilities with detection and remediation patterns. Use when conducting security audits, implementing secure coding practices, or reviewing code for common security vulnerabilities.

Security758 installs

ui-design-aesthetics

nickcrew/claude-cortex

Open the skill to read its SKILL.md instructions.

Frontend & Design358 installs

threat-modeling-techniques

nickcrew/claude-cortex

Open the skill to read its SKILL.md instructions.

Security357 installs

token-efficiency

nickcrew/claude-cortex

Open the skill to read its SKILL.md instructions.

227 installs

helm-chart-patterns

nickcrew/claude-cortex

Open the skill to read its SKILL.md instructions.

DevOps & Cloud179 installs

Popular code quality & workflow skills

View category →

grill-me

mattpocock/skills

A relentless interview to sharpen a plan or design.

Code Quality & Workflow1.4M installs

improve-codebase-architecture

mattpocock/skills

Scan a codebase for deepening opportunities, present them as a visual HTML report, then grill through whichever one you pick.

Code Quality & Workflow1.2M installs

handoff

mattpocock/skills

Compact the current conversation into a handoff document for another agent to pick up.

Code Quality & Workflow1.1M installs

domain-modeling

mattpocock/skills

Build and sharpen a project's domain model. Use when discussing codebase terminology, writing or editing a GLOSSARY.md, or recording or editing an ADR.

Code Quality & Workflow890.6K installs

codebase-design

mattpocock/skills

Shared vocabulary for designing deep modules. Use when the user wants to design or improve a module's interface, find deepening opportunities, decide where a seam goes, make code more testable or AI-navigable, or when another skill needs the deep-module vocabulary.

Code Quality & Workflow863.8K installs

diagnosing-bugs

mattpocock/skills

Diagnosis loop for hard bugs and performance regressions. Use when the user says "diagnose"/"debug this", or reports something broken/throwing/failing/slow.

Code Quality & Workflow848.4K installs

async-python-patterns

Install

Install

Describe the task

Review the result

SKILL.md

Async Python Patterns

When to Use This Skill

Core Patterns

1. Basic Async/Await

2. Concurrent Execution with gather()

3. Task Creation and Management

4. Error Handling

5. Timeout Handling

Advanced Patterns

6. Async Context Managers

7. Async Iterators and Generators

8. Producer-Consumer with Queues

9. Rate Limiting with Semaphores

10. Async Locks for Shared State

Performance Best Practices

1. Use Connection Pools

2. Avoid Blocking the Event Loop

3. Batch Operations

Common Pitfalls

1. Forgetting await

2. Blocking the Event Loop

3. Not Handling Cancellation

4. Mixing Sync and Async

Testing Async Code

Resources

Best Practices Summary

FAQ

More from nickcrew/claude-cortex

Popular code quality & workflow skills