如何使 pydantic 等待异步属性tortoise-orm 的反向外键？

如何解决如何使 pydantic 等待异步属性tortoise-orm 的反向外键？

（问题底部的 MRE）

在 tortoise-orm 中，我们必须像这样等待反向外键字段：

comments = await Post.get(id=id).comments

但是在fastapi中，当返回一个Post实例时，pydantic在抱怨：

pydantic.error_wrappers.ValidationError: 1 validation error for PPost
response -> comments
  value is not a valid list (type=type_error.list)

这是有道理的，因为 comments 属性返回协程。我不得不使用这个小技巧来获得 aronud:

post = Post.get(id=id)
return {**post.__dict__,'comments': await post.comments}

然而，真正的问题是当我有多个关系时：将用户的帖子及其评论返回给用户。在那种情况下，我不得不以一种非常丑陋的方式（这听起来不太好）将我的整个模型转换为 dict。

这是要重现的代码（尽量保持简单）：

models.py

from tortoise.fields import *
from tortoise.models import Model
from tortoise import Tortoise,run_async

async def init_tortoise():
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',modules={'models': ['models']},)
    await Tortoise.generate_schemas()

class User(Model):
    name = CharField(80)

class Post(Model):
    title = CharField(80)
    content = TextField()
    owner = ForeignKeyField('models.User',related_name='posts')

class PostComment(Model):
    text = CharField(80)
    post = ForeignKeyField('models.Post',related_name='comments')

if __name__ == '__main__':
    run_async(init_tortoise())

__all__ = [
    'User','Post','PostComment','init_tortoise',]

ma​​in.py

import asyncio
from typing import List

from fastapi import FastAPI,HTTPException
from pydantic import BaseModel

from models import *


app = FastAPI()

asyncio.create_task(init_tortoise())

# pydantic models are prefixed with P
class PPostComment(BaseModel):
    text: str

class PPost(BaseModel):
    id: int
    title: str
    content: str
    comments: List[PPostComment]
    class Config:
        orm_mode = True

class PUser(BaseModel):
    id: int
    name: str
    posts: List[PPost]
    class Config:
        orm_mode = True

@app.get('/posts/{id}',response_model=PPost)
async def index(id: int):
    post = await Post.get_or_none(id=id)
    return {**post.__dict__,'comments': await post.comments}

@app.get('/users/{id}',response_model=PUser)
async def index(id: int):
    user = await User.get_or_none(id=id)
    return {**user.__dict__,'posts': await user.posts}

/users/1 错误：

pydantic.error_wrappers.ValidationError: 1 validation error for PUser
response -> posts -> 0 -> comments
  value is not a valid list (type=type_error.list)

您也可能希望将其放入 init.py 并运行：

import asyncio
from models import *

async def main():
    await init_tortoise()
    u = await User.create(name='drdilyor')
    p = await Post.create(title='foo',content='lorem ipsum',owner=u)
    c = await PostComment.create(text='spam egg',post=p)

asyncio.run(main())

我想要的是让 pydantic 在这些异步字段上自动等待（这样我就可以返回 Post 实例）。 pydantic 怎么可能？

---

更改/posts/{id}以返回帖子及其所有者而无需评论实际上在使用这种方式时有效（感谢@papple23j）：

    return await Post.get_or_none(id=id).prefetch_related('owner')

但不适用于反向外键。 select_related('comments') 也没有帮助，它正在提高 AttributeError: can't set attribute。

解决方法

（以下文字使用DeepL翻译）

有一种方法可以做到这一点，但有点棘手

首先将pydantic模型片段拆分为schemas.py

from pydantic import BaseModel
from typing import List


# pydantic models are prefixed with P
class PPostComment(BaseModel):
    text: str
    class Config:
        orm_mode = True # add this line

class PPost(BaseModel):
    id: int
    title: str
    content: str
    comments: List[PPostComment]
    class Config:
        orm_mode = True

class PUser(BaseModel):
    id: int
    name: str
    posts: List[PPost]
    class Config:
        orm_mode = True

接下来，重写models.py

from tortoise.fields import *
from tortoise.models import Model
from tortoise import Tortoise,run_async

from schemas import *

async def init_tortoise():
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',modules={'models': ['models']},)
    await Tortoise.generate_schemas()

class User(Model):
    name = CharField(80)
    _posts = ReverseRelation["Post"] #1

    @property
    def posts(self): #3
        return [PPost.from_orm(post) for post in self._posts]

class Post(Model):
    title = CharField(80)
    content = TextField()
    owner = ForeignKeyField('models.User',related_name='_posts') #2
    _comments = ReverseRelation["PostComment"] #1
    
    @property
    def comments(self): #3
        return [PPostComment.from_orm(comment) for comment in self._comments]

class PostComment(Model):
    text = CharField(80)
    post = ForeignKeyField('models.Post',related_name='_comments') #2


if __name__ == '__main__':
    run_async(init_tortoise())

__all__ = [
    'User','Post','PostComment','init_tortoise',]

哪里

#1：使用ReverseRelation来声明反向字段，这里使用底线的前缀来区分

#2：修改 related_name

#3：写一个属性函数，返回对应的pydantic模型列表，这里不需要使用await，因为默认是用prefetch_related()访问它

最后，ma​​in.py

import asyncio
from typing import List

from fastapi import FastAPI,HTTPException

from models import *
from schemas import *
from tortoise.query_utils import Prefetch

app = FastAPI()

asyncio.create_task(init_tortoise())

@app.get('/posts/{id}',response_model=PPost)
async def index(id: int):
    post = await Post.get_or_none(id=id).prefetch_related('_comments') #1
    return PPost.from_orm(post) #2

@app.get('/users/{id}',response_model=PUser)
async def index(id: int):
    user = await User.get_or_none(id=id).prefetch_related(
        Prefetch('_posts',queryset=Post.all().prefetch_related('_comments')) #3
    )
    return PUser.from_orm(user) #2

哪里

#1：使用prefetch_related()预取相关数据

#2：对于带有 orm_mode = True 的乌龟模型，您可以使用 from_orm 将其转换为 pydantic 模型。

#3：对于多层关联数据结构，需要再写一层prefetch_related()

,

您可以尝试使用 prefetch_related()

例如：

@app.get('/posts/{id}',response_model=PPost)
async def index(id: int):
    post = await Post.get_or_none(id=id).prefetch_related('comments')
    return {**post.__dict__}

,

对不起，我太笨了。

我想到的一个解决方案是使用 tortoise.contrib.pydantic 包：

PPost = pydantic_model_creator(Post)
# used as
return await PPost.from_tortoise_orm(await Post.get_or_none(id=1))

但是根据this question，需要在声明模型之前初始化Tortoise，否则不会包含Relation。所以我很想替换这一行：

asyncio.create_task(init_tortoise())

...与：

asyncio.get_event_loop().run_until_complete(init_tortoise())

但它出错了 event loop is already running 并删除了 uvloop 和 installing nest_asyncio helped with that.

---

我使用的解决方案

根据documentation：

可以使用异步和同步接口来获取外键。

异步获取：

events = await tournament.events.all()

同步使用需要在时间之前调用fetch_related，然后才能使用常用函数。

await tournament.fetch_related('events')

在查询集上使用 .fetch_related)（或 prefetch_related）后，反向外键将成为可迭代的，可以用作列表。但是 pydantic 仍然会抱怨这不是有效列表，因此需要使用验证器：

class PPost(BaseModel):
    comments: List[PPostComment]

    @validator('comments',pre=True)
    def _iter_to_list(cls,v):
        return list(v)

（注意验证器不能异步，据我所知）

而且由于我已经设置了 orm_mode，所以我必须使用 .from_orm 方法?：

return PPost.from_orm(await Post.get_or_none(id=42))

请记住，几个小时的反复试验可以为您节省几分钟查看自述文件的时间。