人工智能AI Agent代码智能体多智能体MCP ClientsAgent 编排【免费下载链接】oh-my-openagentOmO: Just type mass ulw keyword with your prompt. Now you are the master of graph engineering.项目地址https://gitcode.com/gh_mirrors/oh/oh-my-openagent点击查看免费下载导读本文是 oh-my-openagent 开源仓库中 programming 技能 所内置的 Python 参考文档《FastAPI SQLAlchemy 2.x async Postgres Pydantic v2》的完整展开。该文档定义了本仓库 Agent 在编写 Python Web API 时必须遵循的“规范栈”canonical stack全链路异步async end-to-end、全链路类型安全type-safe end-to-end、全链路 OpenAPI 自动生成OpenAPI-generated end-to-end。读完本文你将掌握如何用 uv 搭建严格类型检查的项目骨架、如何用 pydantic-settings 管理环境配置、如何用 SQLAlchemy 2.x 的MappedAsDataclass声明式模型配合异步引擎访问 Postgres、如何用 Pydantic v2 分离输入/输出模型、如何用 Alembic 做异步迁移、以及如何用 httpx 的 ASGI 传输层编写不依赖真实数据库的接口测试——同时理解本仓库要求遵守的硬性工程纪律expire_on_commitFalse、禁止 ORM 模型直接作为 API 模型、测试必须走 Given/When/Then 等。一、这套栈在 oh-my-openagent 中的定位与触发方式在 oh-my-openagent 中programming技能位于 packages/shared-skills/skills/programming/SKILL.md是编写.py、.rs、.ts、.go代码前必须加载的“语言门禁”PHASE 0 — LANGUAGE GATE。其 Python 分支要求写任何 Python 代码前必须先读 references/python/README.md再按需加载对应参考文档。当需求涉及 Web API 且需要数据库时就要加载本主题文档 fastapi-stack.md。从 SKILL.md 的“现代生态——标准库选型2026”表可以看到本仓库的硬性选型结论领域Python 规范选择明确禁用Web 框架FastAPIasync、Pydantic 原生、OpenAPIFlask、Django RESTORMSQLAlchemy 2.x asyncMapped[]类型、异步会话Django ORM、Tortoise数据库驱动asyncpg通过 SQLAlchemy其他 PG 驱动数据校验Pydantic v2Rust 核心约比 v1 快 10 倍裸dict边界配置pydantic-settingsenv /.env→ Pydantic 模型零散的os.environ.get(...)包管理uvpip、poetry、conda测试pytest anyiounittestJSON 热路径orjson仅对非 Pydantic 响应热路径上的 stdlibjson这套选型与 references/python/libraries.md 中的决策树一致Web 框架默认 FastAPIORM 默认 SQLAlchemy 2.x async数据库默认 Postgres asyncpg迁移默认 Alembic。本文档就是这些选型在“带数据库的完整 API”场景下的拼装样板。二、项目布局src 布局 迁移目录 测试目录文档给出了一个可直接照抄的目录骨架myapi/ ├── pyproject.toml ├── alembic.ini ├── migrations/ │ └── env.py ├── src/ │ └── myapi/ │ ├── __init__.py │ ├── main.py # FastAPI app lifespan │ ├── config.py # pydantic-settings │ ├── db.py # engine, session factory, dependency │ ├── models.py # SQLAlchemy declarative models │ ├── schemas.py # Pydantic request/response models │ └── routers/ │ ├── __init__.py │ └── users.py └── tests/ ├── conftest.py └── test_users.py要点解读src 布局包代码放在src/myapi/下避免测试与源码相互污染也让uv run uvicorn myapi.main:app的导入路径清晰。关注点分离是强制的config.py配置、db.py基础设施、models.pyORM、schemas.py边界模型、routers/HTTP 层、main.py组装各司其职。这与本仓库 programming 技能“单文件纯代码行数不超过 250 行”的代码异味检测见 SKILL.md 的 CODE SMELLS 章节相互配合——职责一旦超过一个名词短语能概括的范围就必须拆分。测试目录独立tests/conftest.py放夹具tests/test_users.py放接口级测试。三、依赖安装uv 一条命令引入全套使用 uv 添加运行时与开发依赖uv add fastapi sqlalchemy[asyncio]2.0 asyncpg pydantic[email]2 pydantic-settings uvicorn[standard] orjson uv add --dev httpx pytest alembic逐项说明依赖作用注意事项fastapiWeb 框架0.100 即基于 Pydantic v2sqlalchemy[asyncio]2.0异步 ORM必须带asyncioextraasyncpgPostgres 异步驱动URL 协议为postgresqlasyncpg://pydantic[email]2校验 EmailStremailextra 提供 EmailStr 校验器pydantic-settings环境配置加载替代os.environ.getuvicorn[standard]ASGI 服务器standardextra 包含uvloop、httptools等性能组件orjsonJSON 热路径加速必需mandatory原因见下节httpx测试用 ASGI 客户端开发依赖pytest测试框架开发依赖alembic数据库迁移开发依赖为什么orjson被文档标注为“必需”mandatory文档给出的规则是在 FastAPI 应用上设置default_response_classORJSONResponse。其背后的原理与 references/python/orjson-stack.md 一致是Pydantic 类型化响应会绕过 orjson当响应注解为 Pydantic v2 模型时FastAPI 直接调用model_dump_json()该路径由 Rust 编写的 pydantic-core 支撑本身已极快裸dict/list返回值由 ORJSONResponse 接管这部分才能吃到 orjson 相对 stdlibjson的 611 倍序列化提速SSE / NDJSON 流式响应ORJSONResponse会整体缓冲不能用于流式应在StreamingResponse内部对每个 chunk 调用orjson.dumps(...)。关于何时该用 orjson、哪些场景用model_dump_json()更优、以及OPT_NAIVE_UTC/OPT_UTC_Z等标志位参考可继续阅读 orjson-stack.md 的决策树与基准数据。四、配置层pydantic-settings 从环境变量构造类型化配置config.py的规范实现from functools import lru_cache from pydantic import Field, PostgresDsn from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_prefixMYAPI_) database_url: PostgresDsn debug: bool False cors_origins: list[str] Field(default_factorylist) lru_cache def get_settings() - Settings: return Settings() # type: ignore[call-arg] # pydantic populates from env文档在此特意指出一个工程纪律上面代码里的# type: ignore[call-arg]注释违反了本仓库“No type: ignore”的铁律见 README.md 的 iron list —— “Notype: ignore— fix the type error. The checker is right; you are wrong.”。正确做法是为每个字段提供合法默认值让类型检查器无话可说。真正的规范版本class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_prefixMYAPI_) database_url: PostgresDsn debug: bool False cors_origins: list[str] Field(default_factorylist)配置加载的实战细节env_prefixMYAPI_环境变量需以MYAPI_为前缀例如MYAPI_DATABASE_URLpostgresqlasyncpg://...。这是避免多个服务共享 shell 环境时变量名冲突的关键。database_url: PostgresDsn使用 Pydantic 的PostgresDsn类型而非裸str会在边界处校验 URL 格式非法值在启动期直接报错——这正是 README 中“Parse, dont validate”哲学的体现配置在进入程序的那一刻就被解析为类型化值。cors_origins: list[str] Field(default_factorylist)用default_factory而非 []避免可变默认值共享。测试中构造需要指定.env路径时用Settings(_env_file.env)生产环境直接读环境变量即可。lru_cacheget_settings()只在进程生命周期内解析一次配置后续调用直接命中缓存。若测试需要覆盖字段可调用get_settings.cache_clear()后重新获取README 中明确“Pydantic Settings 测试会覆盖字段允许可变”。五、数据库层异步引擎、会话工厂与 FastAPI 依赖注入db.py的规范实现from collections.abc import AsyncIterator from typing import Annotated from fastapi import Depends from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) from myapi.config import get_settings def make_engine() - AsyncEngine: settings get_settings() return create_async_engine( str(settings.database_url), echosettings.debug, pool_pre_pingTrue, ) _engine make_engine() _SessionFactory async_sessionmaker(_engine, expire_on_commitFalse) async def get_session() - AsyncIterator[AsyncSession]: async with _SessionFactory() as session: yield session SessionDep Annotated[AsyncSession, Depends(get_session)]逐层拆解create_async_engine传入str(settings.database_url)PostgresDsn需要转成 str 才能被 SQLAlchemy 接受。echosettings.debug让调试模式下打印 SQL 语句pool_pre_pingTrue在每个连接出借前做一次存活探测避免使用到已被数据库侧断开的陈旧连接——这是“连接池在高负载下被耗尽”类问题的第一道防线。async_sessionmaker(_engine, expire_on_commitFalse)文档特别强调expire_on_commitFalse对 FastAPI 是必需的essential。原因默认expire_on_commitTrue会在 commit 后把实例上的属性标记为“过期”下次访问属性时触发一次隐式 refresh需要回到事件循环重新执行 SQL。在异步环境下这种“同步式隐式加载”会直接抛出MissingGreenlet错误详见下文“常见陷阱”表。get_session生成器依赖async with _SessionFactory() as session: yield session保证每个请求获得独立会话请求结束后会话被正确关闭。依赖的清理语义由 FastAPI 处理——这正是 README 中“资源必须用上下文管理器管理禁止手动.close()”的落点。SessionDep Annotated[AsyncSession, Depends(get_session)]把“依赖注入”声明折叠为一个类型别名。路由函数只需写session: SessionDep即可获得类型完整的异步会话同时 FastAPI 自动完成依赖解析。六、模型层MappedAsDataclass 声明式模型models.py的规范实现from datetime import datetime, UTC from sqlalchemy import DateTime, String, func from sqlalchemy.orm import ( DeclarativeBase, Mapped, MappedAsDataclass, mapped_column, ) class Base(MappedAsDataclass, DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue, initFalse) email: Mapped[str] mapped_column(String(255), uniqueTrue, indexTrue) name: Mapped[str] mapped_column(String(100)) created_at: Mapped[datetime] mapped_column( DateTime(timezoneTrue), server_defaultfunc.now(), initFalse, )两个关键设计MappedAsDataclass让User(email..., name...)成为真正的 dataclass 构造函数字段顺序、默认值、类型注解全部由Mapped[...]声明驱动。ORM 模型天然可变SQLAlchemyMapped[]需要赋值跟踪因此按 README 的规则属于“frozenTrue不适用”的例外代码审查时以# noqa: MUTABLE_OK说明即可。initFalse把由数据库生成的列id自增主键、created_at服务端默认时间戳排除在__init__之外防止调用方误传。server_defaultfunc.now()把时间戳生成下推到数据库保证多进程环境下时间一致DateTime(timezoneTrue)确保存的是带时区的时间对应下文陷阱表中“func.now()返回 naive datetime”的修复。七、Schema 层输入/输出模型必须分离schemas.py的规范实现from datetime import datetime from pydantic import BaseModel, ConfigDict, EmailStr class UserCreate(BaseModel): email: EmailStr name: str class UserRead(BaseModel): model_config ConfigDict(from_attributesTrue) # SQLAlchemy → Pydantic id: int email: EmailStr name: str created_at: datetime文档给出的铁律是始终为同一资源维护独立的*Create输入与*Read输出模型永远不要把 ORM 模型直接暴露为 API 模型。理由安全输入模型只声明客户端可写字段id、created_at这类服务端字段绝不会被客户端注入覆盖。演进未来给UserRead增加计算字段如full_name、别名或json_schema_extra时不会污染写路径。Pydantic v2 迁移语义v1 时代的class Config: orm_mode True已废弃v2 统一使用model_config ConfigDict(from_attributesTrue)。设置后UserRead.model_validate(orm_user)即可把 SQLAlchemy 实例按属性转换为 Pydantic 模型。此外按 libraries.md 的补充输入校验可进一步用Field约束收紧如name: str Field(min_length1, max_length100)并用field_validator在边界完成语义校验。八、路由层类型安全的异步端点routers/users.py的规范实现from fastapi import APIRouter, HTTPException, status from sqlalchemy import select from myapi.db import SessionDep from myapi.models import User from myapi.schemas import UserCreate, UserRead router APIRouter(prefix/users, tags[users]) router.post(, response_modelUserRead, status_codestatus.HTTP_201_CREATED) async def create_user(payload: UserCreate, session: SessionDep) - User: user User(emailpayload.email, namepayload.name) session.add(user) await session.commit() await session.refresh(user) return user router.get(/{user_id}, response_modelUserRead) async def get_user(user_id: int, session: SessionDep) - User: result await session.execute(select(User).where(User.id user_id)) user result.scalar_one_or_none() if user is None: raise HTTPException(status.HTTP_404_NOT_FOUND, User not found) return user router.get(, response_modellist[UserRead]) async def list_users(session: SessionDep, limit: int 100) - list[User]: result await session.execute(select(User).limit(limit)) return list(result.scalars().all())要点response_modelUserRead必须声明声明后 FastAPI 既负责序列化走from_attributesTrue路径也负责生成正确的 OpenAPI schema。若省略response_model直接返回 ORM 对象FastAPI 虽会自动序列化但 OpenAPI 文档会失真——这是陷阱表中明确列出的一项。写操作三步曲session.add(user)→await session.commit()→await session.refresh(user)。refresh从数据库回填id、created_at等initFalse的服务端生成字段返回给客户端的是完整实体。也可以改用await session.flush()await session.refresh(...)组合把 commit 与刷新分离以便在同一事务内做更多操作。读操作select(User).where(User.id user_id)返回Resultscalar_one_or_none()精确表达“零或一条”列表查询用list(result.scalars().all())把Sequence包装成list以满足严格类型检查basedpyrighttypeCheckingMode all下Sequence与list不可混用。路径/查询参数user_id: int由 FastAPI 自动校验类型并生成 OpenAPI 参数定义limit: int 100提供分页上限默认值。九、应用组装lifespan 生命周期管理main.py的规范实现from contextlib import asynccontextmanager from collections.abc import AsyncIterator from fastapi import FastAPI from myapi.config import get_settings from myapi.routers import users asynccontextmanager async def lifespan(_: FastAPI) - AsyncIterator[None]: # Startup: warm up engine pool, run migrations check, etc. yield # Shutdown: close engine from myapi.db import _engine await _engine.dispose() def create_app() - FastAPI: settings get_settings() app FastAPI( titleMy API, debugsettings.debug, lifespanlifespan, ) app.include_router(users.router) return app app create_app()lifespan取代已废弃的app.on_event启动阶段yield之前可做连接池预热、迁移检查、资源初始化关闭阶段yield之后执行await _engine.dispose()优雅回收连接池。这符合 README“资源通过上下文管理器管理同步with/ 异步async with”的硬性规则——asynccontextmanager本身就是资源生命周期管理的内建工具。工厂函数create_app()把应用构建封装成可调用函数便于测试中构造多个隔离实例也便于uvicorn通过myapi.main:app加载模块级单例。启动命令uv run uvicorn myapi.main:app --host 0.0.0.0 --port 8000 --reload--reload仅用于开发生产部署应去掉 reload 并交由进程管理器托管。十、数据库迁移Alembic 异步模板初始化异步迁移模板uv run alembic init -t async migrations-t async会生成使用异步引擎的env.py模板对应 libraries.md 中“Alembic 配置为使用异步引擎”的要求。随后在migrations/env.py中替换target_metadata一行让 autogenerate 能发现模型元数据from myapi.models import Base target_metadata Base.metadata在alembic.ini中设置sqlalchemy.url为异步 URL或更推荐的做法——在env.py中直接复用项目配置保证单一事实来源from myapi.config import get_settings config.set_main_option(sqlalchemy.url, str(get_settings().database_url))生成并应用迁移uv run alembic revision --autogenerate -m create users uv run alembic upgrade head注意alembic revision --autogenerate需要数据库可达autogenerate 会连接数据库对比元数据首次运行前先upgrade head建立alembic_version表或使用alembic revision非 autogenerate手工编写。每次模型变更后都应提交一份新迁移而不是修改已应用的旧迁移。十一、测试httpx ASGI 传输层不启动真实服务器tests/test_users.py的规范实现import pytest from httpx import ASGITransport, AsyncClient from myapi.main import app pytest.mark.anyio async def test_create_and_get_user() - None: async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as client: create_response await client.post( /users, json{email: aliceexample.com, name: Alice}, ) assert create_response.status_code 201 user_id create_response.json()[id] get_response await client.get(f/users/{user_id}) assert get_response.status_code 200 assert get_response.json()[email] aliceexample.com测试设计要点ASGITransport直连 ASGI 应用httpx.AsyncClient配ASGITransport(appapp)会在进程内直接调用 FastAPI 应用无需监听端口、无需真实网络栈。这符合本仓库“更少 mock用真实对象”的测试纪律见 SKILL.md 的 TDD DISCIPLINE——ASGI 传输层是“在线缆层面伪造”但请求处理、Pydantic 校验、依赖注入全部走真实代码。pytest.mark.anyioanyio 的 pytest 插件提供的标记让异步测试运行在 anyio 的后端之上README 硬性规则禁止直接import asyncio异步代码一律基于 anyio。Given/When/Then 结构POST 创建Given 输入 → When 发起请求 → Then 断言 201 与 id 回填再 GET 读取Given 已创建资源 → When 查询 → Then 断言字段一致。一个When对应一次测试关注点。数据库型测试需要真实 Postgres 时在 CI 中用容器跑testcontainers-python或docker-compose并对测试 schema 应用迁移。文档明确警告一旦使用 Postgres 专有类型JSONB、tsvector、数组SQLite 就不能再充当测试替身——类型语义差异会导致测试假绿。这与 SKILL.md 中“testcontainer/sandbox 优于 mock”的层级一致真实 Postgres 测试容器慢但真实。十二、常见陷阱速查表文档以表格形式总结了这套栈最常踩的六个坑全部有明确修复方案陷阱修复commit 后访问关系属性抛出MissingGreenlet在会话工厂上设置expire_on_commitFalse高负载下连接池耗尽在create_async_engine中设置pool_size、max_overflow残留 Pydantic v1 语法orm_mode Truev2 改用model_config ConfigDict(from_attributesTrue)不声明response_model直接返回 ORM 对象声明response_model让 FastAPI 以from_attributesTrue序列化同时保证 OpenAPI 正确await session.execute(...)返回Sequence不满足严格类型用list(result.scalars().all())包装func.now()产生 naive 时间使用DateTime(timezoneTrue)created_at: Mapped[datetime]配UTC感知默认值逐项补充原理MissingGreenlet根因是异步会话在 commit 后对过期属性触发了同步式隐式加载。expire_on_commitFalse让 commit 后实例属性保持原值把“刷新时机”交给显式的await session.refresh(...)。连接池调优create_async_engine支持pool_size池内常驻连接数、max_overflow池满后可超出的临时连接数参数。默认值与业务并发模型不匹配时高并发下会出现“TimeoutError: QueuePool limit of size X overflow Y reached”。结合pool_pre_pingTrue使用可同时规避陈旧连接。Pydantic v1 → v2 迁移class Config内嵌配置在 v2 中被model_config类属性取代orm_mode更名为from_attributes。这是从旧代码迁移到本栈时最常踩的兼容性坑。response_model缺失FastAPI 仍能序列化 ORM 对象Pydantic 会自动做 from-attributes但生成的 OpenAPI schema 会退化为空对象/不完整结构Swagger UI 与客户端代码生成全部失真。Sequencevslistresult.scalars().all()返回Sequence[Row]在 basedpyrighttypeCheckingMode all下与声明为list[...]的返回类型不兼容必须显式list(...)转换——这是“类型系统即证明系统”哲学的日常体现。时区DateTime(timezoneTrue)让列类型感知时区server_defaultfunc.now()配合数据库的timestamptz语义产生带时区时间配合 orjson 的OPT_NAIVE_UTC | OPT_UTC_Z标志见 orjson-stack.md能输出所有解析器都接受的 RFC 3339 时间戳。十三、如何进一步在仓库中深化这套栈编写任何.py文件前先读 SKILL.md 的 PHASE 0 语言门禁再按需加载 Python 参考集。完整阅读 references/python/README.md掌握本仓库的 Python 铁律frozen 默认、NewType品牌类型、match/assert_never穷尽匹配、禁止Any/cast/type: ignore、禁止裸except Exception。配置与测试的扩展范式见 libraries.mdpydantic-settings 完整示例、pytest 的pyproject.toml严格配置。JSON 热路径的完整决策树、选项标志与基准见 orjson-stack.md。异步代码anyio 任务组、取消作用域、通道见 async-anyio.md严格pyproject.tomlbasedpyrighttypeCheckingMode all ruffselect [ALL]见 pyproject-strict.md。这些参考文档由本仓库的oh-my-opencode/shared-skills技能包承载见 packages/shared-skills/AGENTS.md在 OpenCode 与 Codex 双 harness 间共享programming技能的索引路由表位于 SKILL.md 的 Python 跳转表。结语这套“FastAPI SQLAlchemy 2.x async Postgres Pydantic v2”栈的核心竞争力在于三条贯穿始终的主线全链路异步uvicorn → 路由 → 会话 → asyncpg任何一环都不允许阻塞、全链路类型安全Mapped[]注解 ORM、Pydantic 边界校验、response_model声明、SessionDep类型化依赖配合 basedpyright 全量模式、全链路 OpenAPI 自动生成从 Pydantic schema 到路由文档零手写。按本文骨架搭建项目再以expire_on_commitFalse、输入/输出模型分离、容器化 Postgres 测试这三条纪律兜底即可得到一套既快又稳、可直接进入生产迭代的 Python Web API 基线。赞分享人工智能AI Agent代码智能体多智能体MCP ClientsAgent 编排【免费下载链接】oh-my-openagentOmO: Just type mass ulw keyword with your prompt. Now you are the master of graph engineering.项目地址https://gitcode.com/gh_mirrors/oh/oh-my-openagent点击查看免费下载相关推荐FastAPI Expert 实战指南用 Pydantic V2、异步 SQLAlchemy 与 JWT 构建生产级异步 Python APIFastAPI Expert 实战指南用 Pydantic V2、异步 SQLAlchemy 与 JWT 构建生产级异步 Python API 本指南以 clAI 技能AI 插件后端前端DevOps如何快速搭建生产级FastAPI应用完整的DockerPostgres部署指南如何快速搭建生产级FastAPI应用完整的DockerPostgres部署指南 FastAPI生产模板是一个功能全面的项目脚手架专为快速启动生产级Fast输入法弹广告、词库不给力这份 Rime 输入方案清单三步搞定输入法弹广告、词库不给力这份 Rime 输入方案清单三步搞定 你有没有过这样的瞬间输入法右下角弹出购物广告候选词永远排不到你想要的想打一句家乡话却只能蹦人工智能AI Agent代码智能体多智能体MCP ClientsAgent 编排上一篇终极指南如何将Apache MXNet模型无缝导出为ONNX格式下一篇小程序表单开发利器vant-weapp Field组件全功能解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考