如何在FastAPI中结合asyncpg将查询结果映射到Pydantic模型实现校验输出

前置依赖 #

先安装需要的工具包：

pip install fastapi uvicorn asyncpg pydantic

核心原理 #

asyncpg执行查询返回的Record对象是类字典结构，可直接转换为Python字典，传入Pydantic模型的校验方法即可完成数据校验和结构匹配，接口通过指定response_model即可保证最终返回结构完全符合Schema定义。

完整实现示例 #

1. 基础代码编写 #

from datetime import datetime
from typing import Optional, List
from fastapi import FastAPI, Depends, HTTPException
import asyncpg
from pydantic import BaseModel, Field

# 初始化FastAPI实例
app = FastAPI()

# 数据库连接配置，替换为自己的数据库信息
DB_CONFIG = {
    "user": "your_db_user",
    "password": "your_db_pass",
    "database": "your_db_name",
    "host": "127.0.0.1",
    "port": 5432
}

# 数据库连接依赖，自动管理连接释放
async def get_db_conn():
    conn = await asyncpg.connect(**DB_CONFIG)
    try:
        yield conn
    finally:
        await conn.close()

# 定义Pydantic响应Schema，字段名默认和数据库查询返回的列名保持一致
class UserOut(BaseModel):
    id: int
    username: str
    email: str
    create_time: datetime
    # 可选字段标注为Optional，无值时默认返回None
    bio: Optional[str] = None
    # 若数据库列名和Schema字段名不一致，可通过alias指定
    # phone: str = Field(alias="user_phone")

    # Pydantic V2配置，V1版本可删除，改用parse_obj方法校验
    model_config = {
        "from_attributes": True,
        "populate_by_name": True
    }

2. 单条查询接口实现 #

# 指定response_model为定义的Schema，强制返回结构完全匹配定义
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int, conn: asyncpg.Connection = Depends(get_db_conn)):
    # asyncpg使用$1、$2作为参数占位符，避免SQL注入
    row = await conn.fetchrow(
        "SELECT id, username, email, create_time, bio FROM users WHERE id = $1",
        user_id
    )
    if not row:
        raise HTTPException(status_code=404, detail="用户不存在")
    # Record转字典后传入Pydantic完成校验，V1版本替换为UserOut.parse_obj(dict(row))
    return UserOut.model_validate(dict(row))

3. 批量查询接口实现 #

@app.get("/users/", response_model=List[UserOut])
async def list_users(conn: asyncpg.Connection = Depends(get_db_conn)):
    rows = await conn.fetch("SELECT id, username, email, create_time, bio FROM users LIMIT 10")
    # 遍历校验所有查询结果
    return [UserOut.model_validate(dict(row)) for row in rows]

特殊场景处理 #

• 嵌套结构返回：连表查询得到平展结果后，手动组装为符合嵌套Schema结构的字典，再传入Pydantic模型校验即可
• 校验规则扩展：可直接在Pydantic字段上添加校验规则（如字符串长度、数值范围等），查询结果会自动执行校验，不符合规则会直接返回422错误
• 多余字段过滤：Pydantic默认会丢弃Schema未定义的字段，无需手动处理查询返回的冗余列

内容的提问来源于stack exchange，提问作者z3r0