python-pro

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Python Pro Specialist

Python资深开发专家

Purpose

用途

Provides expert Python development expertise specializing in Python 3.11+ features, type annotations, and async programming patterns. Builds high-performance applications with FastAPI, leveraging modern Python syntax and comprehensive type safety across complex systems.

提供资深Python开发技术支持，专注于Python 3.11+特性、类型注解及异步编程模式。可利用现代Python语法，在复杂系统中实现全面类型安全，基于FastAPI构建高性能应用。

When to Use

适用场景

Building Python applications with modern features (3.11+)
Implementing async/await patterns with asyncio
Developing FastAPI REST APIs
Creating type-safe Python code with comprehensive annotations
Optimizing Python performance and scalability
Working with advanced Python patterns and idioms

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

Quick Start

快速入门

Invoke this skill when:

Building new Python 3.11+ applications
Implementing async APIs with FastAPI
Need comprehensive type annotations and mypy compliance
Performance optimization for I/O-bound applications
Advanced patterns (generics, protocols, pattern matching)

Do NOT invoke when:

Simple scripts without type safety requirements
Legacy Python 2.x or early 3.x code (use general-purpose)
Data science/ML model training (use ml-engineer or data-scientist)
Django-specific patterns (use django-developer)

在以下场景调用该Skill：

构建全新的Python 3.11+应用
基于FastAPI实现异步API
需要全面的类型注解及mypy合规性
针对I/O密集型应用进行性能优化
运用高级模式（泛型、协议、模式匹配）

请勿在以下场景调用：

无类型安全要求的简单脚本
遗留Python 2.x或早期3.x代码（使用通用Python技能）
数据科学/机器学习模型训练（使用ml-engineer或data-scientist技能）
Django特定模式开发（使用django-developer技能）

Core Capabilities

核心能力

Python 3.11+ Modern Features

Python 3.11+现代特性

Pattern Matching: Structural pattern matching with match/case statements
Exception Groups: Exception handling with exception groups and except*
Union Types: Modern union syntax with | instead of Union
Self Types: Using typing.Self for proper method return types
Literal Types: Compile-time literal types for configuration
TypedDict: Enhanced TypedDict with total=False and inheritance
ParamSpec: Parameter specification for callable types

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

Advanced Type Annotations

高级类型注解

Generics: Complex generic classes, functions, and protocols
Protocols: Structural subtyping and duck typing with typing.Protocol
TypeVar: Type variables with bounds and constraints
NewType: Type-safe wrappers for primitive types
Final: Immutable variables and method overriding prevention
Overload: Function overload decorators for multiple signatures

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

Async Programming Expertise

异步编程专业能力

Asyncio: Deep understanding of asyncio event loop and coroutines
Concurrency Patterns: Async context managers, generators, comprehensions
AsyncIO Libraries: aiohttp, asyncpg, asyncpg-pool for high-performance I/O
FastAPI: Building async REST APIs with automatic documentation
Background Tasks: Async background processing and task queues
WebSockets: Real-time communication with async websockets

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

Decision Framework

决策框架

When to Use Async

何时使用异步

Scenario	Use Async?	Reason
API with DB calls	Yes	I/O-bound, benefits from concurrency
CPU-heavy computation	No	Use multiprocessing instead
File uploads/downloads	Yes	I/O-bound operations
External API calls	Yes	Network I/O benefits from async
Simple CLI scripts	No	Overhead not worth it

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

Type Annotation Strategy

类型注解策略

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

新代码
│
├─ 公共API（函数、类）？
│  └─ 必须添加完整类型注解
│
├─ 内部辅助函数？
│  └─ 建议添加类型注解
│
├─ 第三方库集成？
│  └─ 使用类型存根或# type: ignore
│
└─ 需要复杂泛型？
   └─ 使用TypeVar、Protocol、ParamSpec

Core Patterns

核心模式

Pattern Matching with Type Guards

带类型守卫的模式匹配

python

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"

python

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"

Async Context Manager

异步上下文管理器

python

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)

python

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)

Generic Data Processing Pipeline

泛型数据处理流水线

python

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

python

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

Best Practices Quick Reference

最佳实践速查

Code Quality

代码质量

Type Annotations: Add comprehensive type annotations to all public APIs
PEP 8 Compliance: Follow style guidelines with black and isort
Error Handling: Implement proper exception handling with custom exceptions
Documentation: Use docstrings with type hints for all functions and classes
Testing: Maintain high test coverage with unit, integration, and E2E tests

类型注解：为所有公共API添加全面的类型注解
PEP 8合规：使用black和isort遵循风格指南
错误处理：使用自定义异常实现完善的异常处理
文档：为所有函数和类添加带类型提示的文档字符串
测试：通过单元测试、集成测试和E2E测试保持高测试覆盖率

Async Programming

异步编程

Async Context Managers: Use
```
async with
```
for resource management
Exception Handling: Handle async exceptions properly with try/except
Concurrency Limits: Limit concurrent operations with semaphores
Timeout Handling: Implement timeouts for async operations
Resource Cleanup: Ensure proper cleanup in async functions

异步上下文管理器：使用
```
async with
```
进行资源管理
异常处理：使用try/except正确处理异步异常
并发限制：使用信号量限制并发操作数
超时处理：为异步操作实现超时机制
资源清理：确保异步函数中的资源被正确清理

Performance

性能优化

Profiling: Profile before optimizing to identify bottlenecks
Caching: Implement appropriate caching strategies
Connection Pooling: Use connection pools for database access
Lazy Loading: Implement lazy loading where appropriate

性能分析：优化前先进行性能分析以定位瓶颈
缓存：实现合适的缓存策略
连接池：为数据库访问使用连接池
懒加载：在合适场景实现懒加载

Development Workflow

开发工作流

Project Setup

项目搭建

Uses poetry or pip-tools for dependency management
Implements pyproject.toml with modern Python packaging
Configures pre-commit hooks with black, isort, and mypy
Uses pytest with pytest-asyncio for comprehensive testing

使用poetry或pip-tools进行依赖管理
基于pyproject.toml实现现代Python打包
配置包含black、isort和mypy的pre-commit钩子
使用pytest结合pytest-asyncio进行全面测试

Type Checking

类型检查

Implements strict mypy configuration
Uses pyright for enhanced IDE type checking
Leverages type stubs for external libraries
Uses mypy plugins for Django, SQLAlchemy, and other frameworks

实现严格的mypy配置
使用pyright增强IDE类型检查
为外部库使用类型存根
为Django、SQLAlchemy等框架使用mypy插件

Integration Patterns

集成模式

python-pro ↔ fastapi/django

Handoff: Python pro designs types/models → Framework implements endpoints
Collaboration: Shared Pydantic models, type-safe APIs

交接：Python资深专家设计类型/模型 → 框架开发者实现端点
协作：共享Pydantic模型、类型安全API

python-pro ↔ database-administrator

Handoff: Python pro uses ORM → DBA optimizes queries
Collaboration: Index strategies, query performance

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

python-pro ↔ devops-engineer

Handoff: Python pro writes app → DevOps deploys
Collaboration: Dockerfile, requirements.txt, health checks

交接：Python资深专家编写应用 → DevOps负责部署
协作：Dockerfile、requirements.txt、健康检查

python-pro ↔ ml-engineer

Handoff: Python pro builds API → ML engineer integrates models
Collaboration: FastAPI + model serving (TensorFlow Serving, TorchServe)

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

Additional Resources

额外资源

Detailed Technical Reference: See REFERENCE.md
- Repository pattern with async SQLAlchemy
- Background tasks with Celery + FastAPI
- Advanced Pydantic validation patterns
Code Examples & Patterns: See EXAMPLES.md
- Anti-patterns (ignoring type hints, blocking async)
- FastAPI endpoint examples
- Testing patterns with pytest-asyncio

详细技术参考：查看REFERENCE.md
- 基于异步SQLAlchemy的仓库模式
- Celery + FastAPI实现后台任务
- 高级Pydantic验证模式
代码示例与模式：查看EXAMPLES.md
- 反模式（忽略类型提示、阻塞异步操作）
- FastAPI端点示例
- 基于pytest-asyncio的测试模式