如何在FastAPI中使用TortoiseORM的ForeignKey关联并正确返回JSON

1 关联字段不展示问题修复 #

默认pydantic_model_creator不会主动包含外键关联对象和反向关联字段，且查询时必须主动预加载关联数据才能正常返回：

1.1 调整Pydantic Schema生成规则 #

在调用pydantic_model_creator时通过include参数指定需要包含的关联字段：

# Toy输出Schema包含关联的owner对象
ToySchema = pydantic_model_creator(Toy, name="Toy", include=("owner",))
# 输入Schema默认已生成外键对应的owner_id字段，无需额外调整
ToySchemaIn = pydantic_model_creator(Toy, name="ToyIn", exclude_readonly=True)

# Person输出Schema包含反向关联的toys_own列表
PersonSchema = pydantic_model_creator(Person, name="Person", include=("toys_own",))
PersonSchemaIn = pydantic_model_creator(Person, name="PersonIn", exclude_readonly=True)

1.2 调整查询接口预加载关联数据 #

查询时使用prefetch_related主动加载关联数据，否则Schema中关联字段会返回空值：

# 查询单个Toy时预加载owner
@app.get("/toy/{toy_id}", response_model=ToySchema)
async def get_toy(toy_id: int):
    return await Toy.get(id=toy_id).prefetch_related("owner")

# 查询所有Toy时预加载owner
@app.get("/toy/", response_model=List[ToySchema])
async def get_all_toys():
    return await Toy.all().prefetch_related("owner")

# 查询单个Person时预加载toys_own
@app.get("/person/{person_id}", response_model=PersonSchema)
async def get_person(person_id: int):
    return await PersonSchema.from_queryset_single(Person.get(id=person_id).prefetch_related("toys_own"))

# 查询所有Person时预加载toys_own
@app.get("/person/", response_model=List[PersonSchema])
async def get_all_persons():
    return await PersonSchema.from_queryset(Person.all().prefetch_related("toys_own"))

2 创建/更新带关联信息的对象 #

2.1 给Toy绑定/修改所属Person #

Tortoise会自动为外键字段生成{外键名}_id的数据库字段，输入Schema默认已包含该字段，直接传值即可：

测试请求示例：创建带所有者的Toy

curl -X 'POST' \
  'http://127.0.0.1:8888/toy' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "乐高积木",
  "owner_id": 1
}'

测试请求示例：修改Toy的所有者

curl -X 'PUT' \
  'http://127.0.0.1:8888/toy/1' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "owner_id": 2
}'

2.2 创建Person时同时创建关联Toy #

需要自定义输入Schema，配合事务实现嵌套创建：

from pydantic import BaseModel
from typing import Optional, List
from tortoise.transactions import in_transaction

# 嵌套创建用的Toy输入Schema
class ToyInForPerson(BaseModel):
    name: str

# 嵌套创建用的Person输入Schema
class PersonWithToyIn(PersonSchemaIn):
    toys: Optional[List[ToyInForPerson]] = None

@app.post("/person-with-toys", response_model=PersonSchema)
async def create_person_with_toys(person: PersonWithToyIn):
    async with in_transaction():
        # 先创建Person对象
        person_obj = await Person.create(name=person.name)
        # 批量创建关联的Toy
        if person.toys:
            toy_objs = [Toy(name=t.name, owner_id=person_obj.id) for t in person.toys]
            await Toy.bulk_create(toy_objs)
    # 返回时预加载关联数据
    return await PersonSchema.from_queryset_single(Person.get(id=person_obj.id).prefetch_related("toys_own"))

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