python-pro

Python Pro Specialist

用途

提供专业的 Python 开发能力，专注于 Python 3.11+ 特性、类型注解和异步编程模式。利用现代 Python 语法和全面的类型安全保障，构建高性能应用，以应对复杂系统的需求。

适用场景

• 使用现代特性（3.11+）构建 Python 应用
• 使用 asyncio 实现 async/await 模式
• 开发 FastAPI REST API
• 使用全面的注解编写类型安全的 Python 代码
• 优化 Python 性能和可扩展性
• 使用高级 Python 模式和惯用写法

快速开始

在以下场景中调用此技能：

• 构建新的 Python 3.11+ 应用
• 使用 FastAPI 实现异步 API
• 需要全面的类型注解并符合 mypy 要求
• 优化 I/O 密集型应用的性能
• 使用高级模式（泛型、协议、模式匹配）

不要在以下场景中调用：

• 不要求类型安全的简单脚本
• 遗留 Python 2.x 或早期 3.x 代码（使用通用技能）
• 数据科学/ML 模型训练（使用 ml-engineer 或 data-scientist）
• Django 专用模式（使用 django-developer）

核心能力

Python 3.11+ 现代特性

• 模式匹配：使用 match/case 语句进行结构化模式匹配
• 异常组：使用异常组和 except* 进行异常处理
• 联合类型：使用 | 代替 Union 的现代联合类型语法
• Self 类型：使用 typing.Self 指定正确的方法返回类型
• 字面量类型：用于配置的编译时字面量类型
• TypedDict：增强的 TypedDict，支持 total=False 和继承
• ParamSpec：为可调用类型指定参数

高级类型注解

• 泛型：复杂的泛型类、函数和协议
• 协议：使用 typing.Protocol 实现结构化子类型和鸭子类型
• TypeVar：带边界和约束的类型变量
• NewType：用于原始类型的类型安全包装器
• Final：不可变变量及防止方法重写
• Overload：用于支持多个签名的函数重载装饰器

异步编程专长

• Asyncio：深入理解 asyncio 事件循环和协程
• 并发模式：异步上下文管理器、生成器和推导式
• AsyncIO 库：aiohttp、asyncpg、asyncpg-pool，适用于高性能 I/O
• FastAPI：构建具备自动文档功能的异步 REST API
• 后台任务：异步后台处理和任务队列
• WebSockets：使用异步 WebSockets 实现实时通信

决策框架

何时使用异步

场景	使用异步？	原因
带数据库调用的 API	是	I/O 密集型，可受益于并发
CPU 密集型计算	否	改用多进程
文件上传/下载	是	I/O 密集型操作
外部 API 调用	是	网络 I/O 可受益于异步
简单的 CLI 脚本	否	开销不值得

类型注解策略

New Code
│
├─ Public API (functions, classes)?
│  └─ Full type annotations required
│
├─ Internal helpers?
│  └─ Type annotations recommended
│
├─ Third-party library integration?
│  └─ Use type stubs or # type: ignore
│
└─ Complex generics needed?
   └─ Use TypeVar, Protocol, ParamSpec

核心模式

使用类型守卫进行模式匹配

from typing import Any

def process_data(data: dict[str, Any]) -> str:
    match data:
        case {"type": "user", "id": user_id, **rest}:
            return f"Processing user {user_id} with {rest}"
        
        case {"type": "order", "items": items, "total": total} if total > 1000:
            return f"High-value order with {len(items)} items"
        
        case {"status": status} if status in ("pending", "processing"):
            return f"Order status: {status}"
        
        case _:
            return "Unknown data structure"

异步上下文管理器

from typing import Optional, Type
from types import TracebackType
import asyncpg

class DatabaseConnection:
    def __init__(self, connection_string: str) -> None:
        self.connection_string = connection_string
        self.connection: Optional[asyncpg.Connection] = None
    
    async def __aenter__(self) -> 'DatabaseConnection':
        self.connection = await asyncpg.connect(self.connection_string)
        return self
    
    async def __aexit__(
        self, 
        exc_type: Optional[Type[BaseException]], 
        exc_val: Optional[BaseException], 
        exc_tb: Optional[TracebackType]
    ) -> None:
        if self.connection:
            await self.connection.close()
    
    async def execute(self, query: str, *args) -> Optional[asyncpg.Record]:
        if not self.connection:
            raise RuntimeError("Connection not established")
        return await self.connection.fetchrow(query, *args)

泛型数据处理流水线

from typing import TypeVar, Generic, Protocol
from abc import ABC, abstractmethod

T = TypeVar('T')
U = TypeVar('U')

class Processor(Protocol[T, U]):
    async def process(self, item: T) -> U: ...

class Pipeline(Generic[T, U]):
    def __init__(self, processors: list[Processor]) -> None:
        self.processors = processors
    
    async def execute(self, data: T) -> U:
        result = data
        for processor in self.processors:
            result = await processor.process(result)
        return result

最佳实践速查

代码质量

• 类型注解：为所有公共 API 添加全面的类型注解
• 符合 PEP 8：使用 black 和 isort 遵循代码风格指南
• 错误处理：通过自定义异常实现恰当的异常处理
• 文档：为所有函数和类编写包含类型提示的文档字符串
• 测试：通过单元测试、集成测试和端到端测试保持较高的测试覆盖率

异步编程

• 异步上下文管理器：使用 async with 管理资源
• 异常处理：使用 try/except 正确处理异步异常
• 并发限制：使用信号量限制并发操作
• 超时处理：为异步操作设置超时
• 资源清理：确保在异步函数中正确清理资源

性能

• 性能分析：先进行性能分析，再优化以找出瓶颈
• 缓存：实施适当的缓存策略
• 连接池：使用连接池访问数据库
• 延迟加载：在适当的场景实施延迟加载

开发工作流

项目设置

• 使用 poetry 或 pip-tools 管理依赖项
• 使用现代 Python 打包方式配置 pyproject.toml
• 使用 black、isort 和 mypy 配置 pre-commit 钩子
• 使用 pytest 和 pytest-asyncio 进行全面测试

类型检查

• 配置严格的 mypy
• 使用 pyright 增强 IDE 类型检查
• 利用外部库的类型存根
• 使用适用于 Django、SQLAlchemy 和其他框架的 mypy 插件

集成模式

python-pro ↔ fastapi/django

• 交接：Python 专家设计类型/模型 → 框架实现端点
• 协作：共享 Pydantic 模型和类型安全的 API

python-pro ↔ database-administrator

• 交接：Python 专家使用 ORM → DBA 优化查询
• 协作：索引策略和查询性能

python-pro ↔ devops-engineer

• 交接：Python 专家编写应用 → DevOps 负责部署
• 协作：Dockerfile、requirements.txt 和运行状况检查

python-pro ↔ ml-engineer

• 交接：Python 专家构建 API → ML 工程师集成模型
• 协作：FastAPI + 模型服务（TensorFlow Serving、TorchServe）

其他资源

• 详细技术参考：参见 REFERENCE.md

  • 使用异步 SQLAlchemy 的仓储模式
  • 使用 Celery + FastAPI 的后台任务
  • 高级 Pydantic 验证模式

• 代码示例与模式：参见 EXAMPLES.md

  • 反模式（忽略类型提示、阻塞异步操作）
  • FastAPI 端点示例
  • 使用 pytest-asyncio 的测试模式