Async Python Patterns

作者 wshobson46891e7e60da無授權條款收錄於 2026年10月8日更新於 2026年10月8日

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.

AI 產生的概覽

提供 Python asyncio、並行程式設計與 async/await 模式的指引,用於非阻塞應用程式。

功能
此技能提供 Python 非同步程式設計的教學指引,涵蓋 asyncio 的事件迴圈、協程、任務、future、非同步情境管理器與非同步迭代器。它說明多種範例模式,包括以 gather() 進行並行執行、建立任務、錯誤處理、逾時與取消,以及常見陷阱和以 pytest-asyncio 進行的測試。它也包含同步與非同步的選擇指南,並指向參考檔案以取得進階模式。
適用情境
適用於建置非同步 Web API、並行 I/O 操作、網路爬蟲、即時服務或其他 I/O 密集的 Python 工作負載。也適合用來判斷是否該採用非同步做法,以及避免常見的 asyncio 錯誤。
執行需求
除代理程式外不需要指令碼或套件;範例涉及 Python 的 asyncio 以及用於測試的 pytest-asyncio,技能另指向 references/details.md 取得進階模式。

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)

來源與署名

來源:wshobson/agents位於plugins/python-development/skills/async-python-patterns提交46891e7

授權條款: 無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架