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.

仅含说明Software Development
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 从公开仓库中收录这些内容。

举报或申请下架

更多来自 wshobson/agents 的技能

Uv Package Manager

wshobson

关于使用 uv Python 包管理器管理依赖、虚拟环境和项目工作流的参考指南。

Software Development2026年10月8日

Web Component Design

wshobson

指导在 React、Vue 和 Svelte 中构建可复用 UI 组件，涵盖组合模式、CSS-in-JS 选择与组件 API 设计。

Software Development2026年10月8日

Python Type Safety

wshobson

指导 Python 类型注解、泛型、协议以及 mypy/pyright 严格类型检查。

Software Development2026年10月8日

Visual Design Foundations

wshobson

指导排版、色彩、间距与图标设计，帮助建立统一且无障碍的视觉设计体系。

Design & Creative2026年10月8日

Responsive Design

wshobson

指导使用容器查询、流体排版、CSS Grid 和断点策略实现响应式网页布局。

Software Development2026年10月8日

Python Resource Management

wshobson

指导 Python 开发者使用上下文管理器、清理模式和流式处理来确定性管理资源。

Software Development2026年10月8日

更多Software Development技能

Playground

anthropics

精选

构建自包含的交互式 HTML 演练场，包含控件、实时预览和可复制的提示词输出。

Software Development37K今天更新

M5 Onboard

anthropics

精选

通过 USB 检测 M5Stack ESP32 开发板，刷写 UIFlow 2.0 固件并安装 MicroPython 应用包。

Software Development37K今天更新

Cardputer Buddy

anthropics

精选

指导迭代 Cardputer-Adv 的 MicroPython 应用包：添加应用、通过 USB 串口推送文件、查看日志并运行 REPL 命令。

Software Development37K今天更新

Testing Strategy

anthropics

精选

设计测试策略与测试计划，涵盖测试类型、覆盖率目标与示例测试用例。

Software Development27K今天更新

Tech Debt

anthropics

精选

识别、分类并排序技术债务，产出评分后的待办清单与分阶段整改计划。

Software Development27K今天更新

System Design

anthropics

精选

通过需求、高层设计、深入设计、扩展与权衡分析来指导系统与架构设计。

Software Development27K今天更新