Async Python Patterns

by wshobson46891e7e60daNo licenseListed Oct 8, 2026Updated Oct 8, 2026

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

Instructions onlySoftware Development
AI-generated overview

Guidance on Python asyncio, concurrency and async/await patterns for non-blocking applications.

What it does
This skill provides instructional guidance on asynchronous Python programming with asyncio, covering coroutines, tasks, futures, async context managers and async iterators. It walks through worked patterns such as concurrent execution with gather(), task creation, error handling, timeouts and cancellation, plus common pitfalls and testing with pytest-asyncio. It also includes a sync-versus-async decision guide and points to a reference file for advanced patterns.
When to use it
Use it when building async web APIs, concurrent I/O operations, web scrapers, real-time services or other I/O-bound workloads in Python. It is also useful for deciding whether async is appropriate and for avoiding typical asyncio mistakes.
Requirements
No scripts or packages are required beyond an agent; examples reference Python's asyncio and pytest-asyncio for testing, and the skill points to references/details.md for advanced patterns.

Async Python Patterns

Comprehensive 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
  • Building microservices with async communication
  • Optimizing I/O-bound workloads
  • Implementing async background tasks and queues

Sync vs Async Decision Guide

Before adopting async, consider whether it's the right choice for your use case.

Use CaseRecommended Approach
Many concurrent network/DB callsasyncio
CPU-bound computationmultiprocessing or thread pool
Mixed I/O + CPUOffload CPU work with asyncio.to_thread()
Simple scripts, few connectionsSync (simpler, easier to debug)
Web APIs with high concurrencyAsync frameworks (FastAPI, aiohttp)

Key Rule: Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity.

Core Concepts

1. Event Loop

The event loop is the heart of asyncio, managing and scheduling asynchronous tasks.

Key characteristics:

  • Single-threaded cooperative multitasking
  • Schedules coroutines for execution
  • Handles I/O operations without blocking
  • Manages callbacks and futures

2. Coroutines

Functions defined with async def that can be paused and resumed.

Syntax:

python
async def my_coroutine():    result = await some_async_operation()    return result

3. Tasks

Scheduled coroutines that run concurrently on the event loop.

4. Futures

Low-level objects representing eventual results of async operations.

5. Async Context Managers

Resources that support async with for proper cleanup.

6. Async Iterators

Objects that support async for for iterating over async data sources.

Quick Start

python
import asyncio
async def main():    print("Hello")    await asyncio.sleep(1)    print("World")
# Python 3.7+asyncio.run(main())

Fundamental Patterns

Pattern 1: Basic Async/Await

python
import asyncio
async def fetch_data(url: str) -> dict:    """Fetch data from URL 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())

Pattern 2: Concurrent Execution with gather()

python
import asynciofrom typing import List
async def fetch_user(user_id: int) -> dict:    """Fetch user data."""    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
async def main():    user_ids = [1, 2, 3, 4, 5]    users = await fetch_all_users(user_ids)    print(f"Fetched {len(users)} users")
asyncio.run(main())

Pattern 3: Task Creation and Management

python
import asyncio
async def background_task(name: str, delay: int):    """Long-running background task."""    print(f"{name} started")    await asyncio.sleep(delay)    print(f"{name} completed")    return f"Result from {name}"
async def main():    # Create tasks    task1 = asyncio.create_task(background_task("Task 1", 2))    task2 = asyncio.create_task(background_task("Task 2", 1))
    # Do other work    print("Main: doing other work")    await asyncio.sleep(0.5)
    # Wait for tasks    result1 = await task1    result2 = await task2
    print(f"Results: {result1}, {result2}")
asyncio.run(main())

Pattern 4: Error Handling in Async Code

python
import asynciofrom typing import List, Optional
async def risky_operation(item_id: int) -> dict:    """Operation that might fail."""    await asyncio.sleep(0.1)    if item_id % 3 == 0:        raise ValueError(f"Item {item_id} failed")    return {"id": item_id, "status": "success"}
async def safe_operation(item_id: int) -> Optional[dict]:    """Wrapper with error handling."""    try:        return await risky_operation(item_id)    except ValueError as e:        print(f"Error: {e}")        return None
async def process_items(item_ids: List[int]):    """Process multiple items with error handling."""    tasks = [safe_operation(iid) for iid in item_ids]    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Filter out failures    successful = [r for r in results if r is not None 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
asyncio.run(process_items([1, 2, 3, 4, 5, 6]))

Pattern 5: Timeout Handling

python
import asyncio
async def slow_operation(delay: int) -> str:    """Operation that takes time."""    await asyncio.sleep(delay)    return f"Completed after {delay}s"
async def with_timeout():    """Execute operation with timeout."""    try:        result = await asyncio.wait_for(slow_operation(5), timeout=2.0)        print(result)    except asyncio.TimeoutError:        print("Operation timed out")
asyncio.run(with_timeout())

Detailed worked examples and patterns

Detailed sections (starting with ## Advanced Patterns) live in references/details.md. Read that file when the navigation summary above is insufficient.

Common Pitfalls

1. Forgetting await

python
# Wrong - returns coroutine object, doesn't executeresult = async_function()
# Correctresult = await async_function()

2. Blocking the Event Loop

python
# Wrong - blocks event loopimport timeasync def bad():    time.sleep(1)  # Blocks!
# Correctasync def good():    await asyncio.sleep(1)  # Non-blocking

3. Not Handling Cancellation

python
async def cancelable_task():    """Task that handles cancellation."""    try:        while True:            await asyncio.sleep(1)            print("Working...")    except asyncio.CancelledError:        print("Task cancelled, cleaning up...")        # Perform cleanup        raise  # Re-raise to propagate cancellation

4. Mixing Sync and Async Code

python
# Wrong - can't call async from sync directlydef sync_function():    result = await async_function()  # SyntaxError!
# Correctdef sync_function():    result = asyncio.run(async_function())

Testing Async Code

python
import asyncioimport pytest
# Using pytest-asyncio@pytest.mark.asyncioasync def test_async_function():    """Test async function."""    result = await fetch_data("https://api.example.com")    assert result is not None
@pytest.mark.asyncioasync def test_with_timeout():    """Test with timeout."""    with pytest.raises(asyncio.TimeoutError):        await asyncio.wait_for(slow_operation(5), timeout=1.0)

Source and attribution

Source:wshobson/agentsinplugins/python-development/skills/async-python-patternsat commit46891e7

License: No license

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

More from wshobson/agents

Uv Package Manager

wshobson

Reference guide for using the uv Python package manager for dependencies, virtual environments and project workflows.

Software DevelopmentOct 8, 2026

Web Component Design

wshobson

Guides building reusable UI components in React, Vue and Svelte, covering composition patterns, CSS-in-JS choices and component API design.

Software DevelopmentOct 8, 2026

Python Type Safety

wshobson

Guides Python type annotations, generics, protocols, and strict mypy/pyright checking.

Software DevelopmentOct 8, 2026

Visual Design Foundations

wshobson

Guides typography, color, spacing and iconography decisions for cohesive, accessible visual design systems.

Design & CreativeOct 8, 2026

Responsive Design

wshobson

Guides implementation of responsive web layouts using container queries, fluid typography, CSS Grid and breakpoints.

Software DevelopmentOct 8, 2026

Python Resource Management

wshobson

Guides Python developers in managing resources deterministically with context managers, cleanup patterns, and streaming.

Software DevelopmentOct 8, 2026

More in Software Development

Playground

anthropics

Featured

Builds self-contained interactive HTML playgrounds with controls, live preview, and copyable prompt output.

Software Development37Kupdated today

M5 Onboard

anthropics

Featured

Provisions M5Stack ESP32 boards by detecting them on USB, flashing UIFlow 2.0 firmware, and installing a MicroPython app bundle.

Software Development37Kupdated today

Cardputer Buddy

anthropics

Featured

Guides iterating on a Cardputer-Adv MicroPython app bundle: adding apps, pushing files over USB-serial, tailing logs, and running REPL commands.

Software Development37Kupdated today

Dpop Adoption

google

Featured

Guides implementation of OAuth 2.0 DPoP (RFC 9449) sender-constrained refresh tokens for Google's OAuth platform.

Software Development21Kupdated today

Gemini Live Api

google

Featured

Generates a Gemini Live API WebSocket client class, tests, docs, and a demo UI for bidirectional streaming.

Software Development21Kupdated today

Gemini Api

google

Featured

Guides use of the Gemini API on Google's Agent Platform with the Gen AI SDK across languages and capabilities.

Software Development21Kupdated today