moai-lang-python

快速参考（30 秒）

Python 3.13 及以上版本开发专家——FastAPI、Django、异步模式、pytest 和现代 Python 特性。

自动触发：扩展名为 .py 的 Python 文件、pyproject.toml、requirements.txt、pytest.ini，以及有关 FastAPI 或 Django 的讨论

核心能力：

• Python 3.13 特性：通过 PEP 744 使用 JIT 编译器、通过 PEP 703 使用无 GIL 模式，以及使用 match 和 case 语句进行模式匹配
• Web 框架：FastAPI 0.115 及更高版本、Django 5.2 LTS
• 数据验证：Pydantic v2.9 的 model_validate 模式
• ORM：SQLAlchemy 2.0 异步模式
• 测试：使用 pytest 的夹具、异步测试和 parametrize 装饰器
• 包管理：使用 pyproject.toml 的 poetry、uv、pip
• 类型提示：Protocol、TypeVar、ParamSpec 和现代类型模式
• 异步：asyncio、异步生成器和任务组
• 数据科学：numpy、pandas 和 polars 基础知识

快速模式

FastAPI 端点模式：

从 fastapi 导入 FastAPI 和 Depends，并从 pydantic 导入 BaseModel。创建一个 FastAPI 应用实例。定义一个继承自 BaseModel 的 UserCreate 模型类，其中包含 name 和 email 字符串字段。在 users 路径创建一个异步 post 端点，接收一个 UserCreate 参数，并通过 await 调用 UserService.create 返回一个 User。

Pydantic v2.9 验证模式：

从 pydantic 导入 BaseModel 和 ConfigDict。定义一个继承自 BaseModel 的 User 类。使用 ConfigDict 设置 model_config，将 from_attributes 设为 True，并将 str_strip_whitespace 设为 True。添加 id 整数字段、name 字符串字段和 email 字符串字段。使用 model_validate 从 ORM 对象创建实例，使用 model_validate_json 从 JSON 数据创建实例。

pytest 异步测试模式：

导入 pytest，并使用 pytest.mark.asyncio 装饰器标记测试函数。创建一个异步测试函数，并将 async_client 作为夹具参数传入。向 users 端点发送 post 请求，并附带包含 name 字段的 JSON 请求体。断言响应的 status_code 等于 201。

---

实施指南（5 分钟）

Python 3.13 新特性

通过 PEP 744 使用 JIT 编译器：

• 实验性功能，默认禁用
• 将 PYTHON_JIT 环境变量设为 1 即可启用
• 可通过 enable-experimental-jit 构建选项启用
• 可提升 CPU 密集型代码的性能
• 使用 copy-and-patch JIT，将专用字节码转换为机器码

通过 PEP 703 使用无 GIL 模式：

• 可用的实验性自由线程构建版本为 python3.13t
• 支持真正的线程并行执行
• 可通过官方 Windows 和 macOS 安装程序获取
• 最适用于 CPU 密集型多线程应用
• 目前不建议用于生产环境

使用 match 和 case 进行模式匹配：

创建一个 process_response 函数，接收一个响应字典并返回一个字符串。对 response 使用 match 语句。对于包含 status 为 ok 且带有 data 字段的 case，返回包含该数据的成功消息。对于包含 status 为 error 且带有 message 字段的 case，返回错误消息。对于 status 匹配 pending 或 processing 且满足守卫条件的 case，返回进行中消息。对于使用下划线的默认 case，返回未知响应。

FastAPI 0.115+ 模式

异步依赖注入：

从 fastapi 导入 FastAPI、Depends，从 sqlalchemy.ext.asyncio 导入 AsyncSession，并从 contextlib 导入 asynccontextmanager。创建一个用 asynccontextmanager 装饰的 lifespan 异步上下文管理器，并让它接收 FastAPI 应用。在 lifespan 中，启动时调用 await init_db，然后 yield，最后在关闭时调用 await cleanup。创建 FastAPI 应用时传入 lifespan 参数。定义一个异步 get_db 函数，返回 AsyncSession 的 AsyncGenerator，并使用 async with 处理 async_session，然后 yield 该会话。创建一个用于获取用户的 get 端点，包含 user_id 路径参数，并使用 Depends 和 get_db 注入数据库会话。调用 await get_user_by_id，并用该用户返回 UserResponse.model_validate。

基于类的依赖：

创建一个 Paginator 类，其 init 方法接受 page（默认值为 1）和 size（默认值为 20）。将 self.page 设为 1 和 page 中的较大值，将 self.size 设为 100 和 1 与 size 中较大值的较小值，并将 self.offset 设为 page 减 1 后乘以 size。创建一个 list_items 端点，使用 Depends 注入 Paginator 以获取分页参数，并使用 get_page 和 offset、size 返回条目。

Django 5.2 LTS 特性

复合主键：

创建一个 OrderItem 模型，其中包含一个指向 Order、删除时使用 CASCADE 的 ForeignKey，一个指向 Product、删除时使用 CASCADE 的 ForeignKey，以及一个用于 quantity 的 IntegerField。在 Meta 类中，将 pk 设为包含 order 和 product 字段的 models.CompositePrimaryKey。

带查询参数的 URL 反向解析：

从 django.urls 导入 reverse。调用 reverse，传入搜索视图名称、包含 q 设为 django 且 page 设为 1 的查询字典，以及设为 results 的 fragment。结果是包含查询字符串和片段的搜索路径。

Shell 中的模型自动导入：

运行 python manage.py shell，所有已安装应用中的模型都会自动导入，无需显式导入语句。

Pydantic v2.9 深度模式

使用 Annotated 创建可复用的验证器：

从 typing 导入 Annotated，从 pydantic 导入 AfterValidator 和 BaseModel。定义一个 validate_positive 函数，接收整数 v 并返回整数。如果 v 小于或等于 0，则抛出错误消息为 must be positive 的 ValueError。否则返回 v。将 PositiveInt 创建为由 int 和使用 validate_positive 的 AfterValidator 构成的 Annotated。模型字段 price 和 quantity 使用 PositiveInt。

使用模型验证器进行跨字段验证：

从 pydantic 导入 BaseModel 和 model_validator，并从 typing 导入 Self。创建一个 DateRange 模型，其中 start_date 和 end_date 为 date 字段。添加一个 model_validator 装饰器，并将 mode 设为 after。在返回 Self 的 validate_dates 方法中，检查 end_date 是否早于 start_date；如果是，则抛出 ValueError，否则返回 self。

ConfigDict 最佳实践：

创建一个 BaseSchema 模型，并将 model_config 设为 ConfigDict。将 from_attributes 设为 True，以支持 ORM 对象；将 populate_by_name 设为 True，以允许使用别名；将 extra 设为 forbid，以拒绝未知字段；将 str_strip_whitespace 设为 True，以清理字符串。

SQLAlchemy 2.0 异步模式

引擎和会话设置：

从 sqlalchemy.ext.asyncio 导入 create_async_engine、async_sessionmaker 和 AsyncSession。使用 create_async_engine 和 postgresql+asyncpg 连接字符串创建引擎，将 pool_pre_ping 设为 True，并将 echo 设为 True。使用 async_sessionmaker 创建 async_session，传入引擎，将 class_ 设为 AsyncSession，并将 expire_on_commit 设为 False，以防止出现分离实例错误。

仓储模式：

创建一个 UserRepository 类，其 init 方法接收一个 AsyncSession。定义一个异步 get_by_id 方法，执行带有 user_id where 子句的 select 查询，并返回 scalar_one_or_none 结果。定义一个异步 create 方法，从 UserCreate model_dump 创建一个 User，将其添加到会话中，提交、刷新并返回该用户。

流式处理大型结果：

创建一个接收 AsyncSession 的异步 stream_users 函数。调用 await db.stream 和 select User 查询。使用 async for 遍历 result.scalars 并 yield 每个用户。

pytest 高级模式

使用 pytest-asyncio 的异步 fixture：

从 httpx 导入 pytest、pytest_asyncio 和 AsyncClient。使用 pytest_asyncio.fixture 装饰 fixture。创建一个 async_client fixture，在 AsyncClient 上使用 async with，并传入 app 和 base_url，然后 yield 客户端。创建一个 db_session fixture，在 async_session 和 session.begin 上使用 async with，yield 会话并调用 await session.rollback。

参数化测试：

使用 pytest.mark.parametrize 装饰器，参数名为 input_data 和 expected_status。提供由字典和预期状态码组成的元组作为测试用例。为 valid、empty_name 和 missing_name 用例添加 ids。测试函数接收 async_client、input_data 和 expected_status，向 users 端点发起请求，并断言 status_code 与预期值一致。

Fixture 工厂：

创建一个 user_factory fixture，返回一个异步函数。内部函数接收 db（类型为 AsyncSession）和关键字参数。设置包含 name 和 email 的默认值字典。使用管道运算符将默认值与 kwargs 合并来创建 User，将其添加到 db，提交并返回该用户。

类型提示的现代模式

用于结构化类型的 Protocol：

从 typing 导入 Protocol 和 runtime_checkable。应用 runtime_checkable 装饰器。定义一个带有泛型类型 T 的 Repository Protocol。添加抽象异步 get 方法，接收 int 类型的 id 并返回 T 或 None；添加异步 create 方法，接收 dict 类型的 data 并返回 T；添加异步 delete 方法，接收 int 类型的 id 并返回 bool。

用于装饰器的 ParamSpec：

从 typing 导入 ParamSpec、TypeVar 和 Callable，并从 functools 导入 wraps。将 P 定义为 ParamSpec，将 R 定义为 TypeVar。创建一个 retry 装饰器函数，接收默认值为 3 的 times，并返回一个可调用的包装器。内部装饰器使用 wraps 包装函数，包装器按指定的次数循环，尝试 await 该函数，并在最后一次尝试时重新抛出异常。

包管理

使用 Poetry 的 pyproject.toml：

在 tool.poetry 部分中，设置名称、版本和 Python 版本约束。在 dependencies 下添加 fastapi、pydantic 和带有 asyncio extra 的 sqlalchemy。在 dev dependencies 下添加 pytest、pytest-asyncio 和 ruff。配置 ruff 的 line-length 和 target-version。在 ini_options 中将 pytest asyncio_mode 设为 auto。

快速包管理器 uv：

使用来自 astral.sh 的安装脚本，通过 curl 安装 uv。使用 uv venv 创建虚拟环境。使用 uv pip install 从 requirements.txt 安装依赖。使用 uv add 命令添加依赖。

---

高级实现（10 分钟以上）

如需全面了解以下内容：

• Docker 和 Kubernetes 的生产部署模式
• 高级异步模式，包括任务组和信号量
• 使用 numpy、pandas 和 polars 进行数据科学集成
• 性能优化技术
• 遵循 OWASP 模式的安全最佳实践
• CI/CD 集成模式

请参阅：

• reference.md：完整参考文档
• examples.md：可用于生产环境的代码示例

---

Context7 库映射

• tiangolo/fastapi：FastAPI 异步 Web 框架
• django/django：Django Web 框架
• pydantic/pydantic：使用类型注解进行数据验证
• sqlalchemy/sqlalchemy：SQL 工具包和 ORM
• pytest-dev/pytest：测试框架
• numpy/numpy：数值计算
• pandas-dev/pandas：数据分析库
• pola-rs/polars：快速 DataFrame 库

---

可搭配使用

• moai-domain-backend：REST API 和微服务架构
• moai-domain-database：SQL 模式和 ORM 优化
• moai-workflow-testing：DDD 和测试策略
• moai-essentials-debug：AI 驱动的调试
• moai-foundation-quality：TRUST 5 质量原则

---

故障排除

常见问题：

Python 版本检查：

使用带有版本标志的 python 运行命令，确认版本为 3.13 或更高。使用带有 -c 标志的 python 输出 sys.version_info，以获取详细的版本信息。

异步会话分离错误：

在会话配置中将 expire_on_commit 设为 False。或者，在提交后对对象调用 await session.refresh。

pytest asyncio 模式警告：

在 pyproject.toml 的 tool.pytest.ini_options 下，将 asyncio_mode 设为 auto，并将 asyncio_default_fixture_loop_scope 设为 function。

Pydantic v2 迁移：

parse_obj 方法现为 model_validate。parse_raw 方法现为 model_validate_json。from_orm 功能要求在 ConfigDict 中将 from_attributes 设为 True。

---

最后更新：2026-01-11 状态：活跃（v1.1.0）