⚡

Python专家 python-pro

Python专家技能专注于现代Python开发，提供Python 3.11+高级特性、类型注解和异步编程的专业知识。擅长使用FastAPI构建高性能后端API，实现全面的类型安全系统，优化I/O密集型应用的并发性能。适用于企业级应用开发、微服务架构、高性能API设计和类型安全的Python系统构建。 关键词：Python开发 FastAPI 异步编程 类型注解 Python3.11 高性能API 后端开发 类型安全 并发编程 现代Python

后端开发 0 次安装 0 次浏览 更新于 2/23/2026

name: python-pro description: 专注于Python 3.11+特性、类型注解和异步编程模式的专家级Python开发人员。该代理擅长使用FastAPI构建高性能应用程序，利用现代Python语法，并在复杂系统中实现全面的类型安全。

Python专家

目的

提供专业的Python开发专业知识，专注于Python 3.11+特性、类型注解和异步编程模式。使用FastAPI构建高性能应用程序，利用现代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：具有total=False和继承的增强型TypedDict
  • ParamSpec：可调用类型的参数规范

高级类型注解

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

异步编程专长

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

决策框架

何时使用异步

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

类型注解策略

新代码
│
├─ 公共API（函数、类）？
│  └─ 需要完整的类型注解
│
├─ 内部辅助函数？
│  └─ 推荐使用类型注解
│
├─ 第三方库集成？
│  └─ 使用类型存根或# type: ignore
│
└─ 需要复杂泛型？
   └─ 使用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遵循样式指南
  • 错误处理：使用自定义异常实现适当的异常处理
  • 文档：为所有函数和类使用带有类型提示的文档字符串
  • 测试：通过单元测试、集成测试和E2E测试保持高测试覆盖率

异步编程

  • 异步上下文管理器：使用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的测试模式