技能 async-python-patterns
🐍

async-python-patterns

安全

实现异步Python模式

也可从以下获取: wshobson,ActiveInferenceInstitute

使用asyncio模式构建高性能Python应用,实现并发I/O操作、异步API和非阻塞系统。

支持: Claude Codex Code(CC)
🥉 75 青铜
1

下载技能 ZIP

2

在 Claude 中上传

前往 设置 → 功能 → 技能 → 上传技能

3

开启并开始使用

测试它

正在使用“async-python-patterns”。 如何在Python中并发运行多个异步任务?

预期结果:

使用asyncio.gather()并发运行多个协程。示例:results = await asyncio.gather(task1(), task2(), task3())。要进行错误处理,请使用return_exceptions=True来捕获失败而不停止其他任务。

正在使用“async-python-patterns”。 asyncio.create_task和asyncio.ensure_future有什么区别?

预期结果:

create_task()调度一个协程在事件循环上运行并返回一个Task(Python 3.7+)。ensure_future()接受协程和future,返回Task。对新协程使用create_task();当您可能收到现有future时使用ensure_future()。

正在使用“async-python-patterns”。 如何为异步操作添加超时?

预期结果:

使用asyncio.wait_for()进行简单超时:await asyncio.wait_for(async_func(), timeout=5.0)。对于多个操作,请使用带超时的asyncio.wait()并在try/except块中处理'timeout'异常。

安全审计

安全
v1 • 2/24/2026

This skill is a documentation and educational resource for async Python programming patterns. The static analyzer flagged 71 potential issues, but upon manual review, all findings are FALSE POSITIVES. The 'external_commands' detections are markdown code fences (backticks), 'network' detections are example URLs in documentation, and 'blocker' detections are benign documentation patterns. No actual security risks were found.

2
已扫描文件
721
分析行数
0
发现项
1
审计总数
未发现安全问题

检测到的模式

Markdown Code Fence Detection (False Positive)Example URLs in Documentation (False Positive)Documentation Pattern Detection (False Positive)Cryptographic Algorithm Detection (False Positive)
审计者: claude

质量评分

38
架构
100
可维护性
87
内容
50
社区
100
安全
100
规范符合性

你能构建什么

构建异步Web API

使用FastAPI创建高性能REST API,包含并发数据库查询和异步端点

并发数据获取

使用asyncio.gather同时获取多个API资源,包含正确的错误处理和超时

实时应用后端

实现WebSocket服务器和异步消息处理,用于聊天或实时通知系统

试试这些提示

基本异步模式请求
向我展示如何在Python中编写一个简单的异步函数来从URL获取数据。包含正确的错误处理和超时。
并发操作
我需要同时从5个不同的API端点获取数据。展示如何使用asyncio.gather和return_exceptions=True进行错误处理。
异步上下文管理器
创建一个异步上下文管理器,连接到数据库,yield一个连接,并在退出时正确关闭它。包含错误处理。
生产级异步模式
构建一个生产级异步函数,使用连接池查询数据库,实现指数退避的重试逻辑,包含结构化日志,并优雅地处理取消。

最佳实践

  • 始终添加超时以防止无限阻塞——使用asyncio.wait_for()或asyncio.timeout()
  • 使用Python 3.11+中的结构化并发asyncio.TaskGroup进行自动清理
  • 在async函数内部显式处理异常,使用try/except,切勿让异常静默传播

避免

  • 不要在异步函数内使用阻塞同步调用——这会阻塞整个事件循环
  • 避免'即发即忘'而没有正确的任务跟踪——存储Task引用以启用取消
  • 不要混合await和阻塞代码如time.sleep()——改用asyncio.sleep()

常见问题

什么时候应该使用asyncio vs线程?
对I/O密集型任务(网络、文件、数据库)使用asyncio,在等待占主导的地方使用。对需要真正并行性的CPU密集型任务使用线程。Asyncio以比线程更少的内存处理数千个并发I/O操作。
如何在异步上下文中运行同步阻塞代码?
使用asyncio.to_thread()(Python 3.9+)或loop.run_in_executor()在线程池中运行阻塞同步代码而不阻塞事件循环。
gather、create_task和wait有什么区别?
asyncio.gather()运行所有协程并一起返回所有结果。create_task()调度单个协程并返回Task以供稍后等待。asyncio.wait()返回done/pending集合而不收集结果。
如何处理并发任务中的异常?
在gather()中使用return_exceptions=True将所有异常捕获为结果。对于结构化并发,使用TaskGroup(Python 3.11+)自动捕获异常并在失败时取消其他任务。
我可以在async函数外部使用async with吗?
不,async with必须在async函数内部。在顶层用asyncio.run()包装它,或在Python 3.10+中使用asyncio.current_task()使用运行中的事件循环。
如何有效调试异步代码?
使用PYTHONASYNCIODEBUG=1环境变量进行警告。使用asyncio.get_running_loop().set_debug(True)获取详细堆栈跟踪。安装aiomonitor或使用asyncio.sleep()断点而不是常规断点。