1

(问题底部的 MRE)

在 tortoise-orm 中,我们必须等待反向 ForeignKey 字段,如下所示:

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(这听起来不太好)。

这是要重现的代码(试图使其尽可能简单):

模型.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',
]

主文件

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

4

3 回答 3

1

(以下文字使用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()

最后,main.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()

于 2021-05-05T07:15:43.507 回答
0

您可以尝试使用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__}
于 2021-04-30T17:14:13.070 回答
0

对不起,我太笨了。

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

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

但是根据这个问题,需要在声明模型之前初始化 Tortoise,否则不会包含关系。所以我很想替换这一行:

asyncio.create_task(init_tortoise())

...和:

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

但它出错了event loop is already running,删除了 uvloop 并安装了 nest_asyncio 有助于解决这个问题。


我使用的解决方案

根据文档

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

异步获取:

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))

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

于 2021-06-07T16:46:01.683 回答