如何让 pydantic await 异步 属性（tortoise-orm 的反向外键）？
How to make pydantic await on a async property (tortoise-orm's reverse ForeignKey)?
(MRE在题底)
在 tortoise-orm 中，我们必须像这样等待反向外键字段：
comments = await Post.get(id=id).comments
但是在 fastapi 中，当 returning 一个 Post 实例时，pydantic 抱怨：
pydantic.error_wrappers.ValidationError: 1 validation error for PPost
response -> comments
value is not a valid list (type=type_error.list)
作为 comments 属性 return 协程是有意义的。我不得不使用这个小技巧来获得 aronud:
post = Post.get(id=id)
return {**post.__dict__, 'comments': await post.comments}
但是，真正的问题 是当我有多个关系时：return 一个用户及其 post 及其评论。在那种情况下，我不得不以一种非常丑陋的方式（听起来不太好）将我的整个模型转换为 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',
]
main.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 在那些异步字段上自动等待（这样我就可以 return Post 实例）。 pydantic 怎么可能？
将 /posts/{id} 更改为 return post 及其 owner 没有注释实际上是使用这种方式工作（感谢@papple23j）：
return await Post.get_or_none(id=id).prefetch_related('owner')
但不适用于反向外键。 select_related('comments') 也没有帮助，它正在提高 AttributeError: can't set attribute.
您可以尝试使用 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__}
（以下文字使用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：写一个属性函数和return对应的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()
对不起，我太笨了
我想到的一个解决方案是使用 tortoise.contrib.pydantic 包：
PPost = pydantic_model_creator(Post)
# used as
return await PPost.from_tortoise_orm(await Post.get_or_none(id=1))
但是根据，在声明模型之前需要初始化 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.
我使用的解决方案
Fetching foreign keys can be done with both async and sync interfaces.
Async fetch:
events = await tournament.events.all()
Sync usage requires that you call fetch_related before the time, and then you can use common functions.
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))
Remember, a few hours of trial and error can save you several minutes of looking at the README.
(MRE在题底)
在 tortoise-orm 中，我们必须像这样等待反向外键字段：
comments = await Post.get(id=id).comments
但是在 fastapi 中，当 returning 一个 Post 实例时，pydantic 抱怨：
pydantic.error_wrappers.ValidationError: 1 validation error for PPost
response -> comments
value is not a valid list (type=type_error.list)
作为 comments 属性 return 协程是有意义的。我不得不使用这个小技巧来获得 aronud:
post = Post.get(id=id)
return {**post.__dict__, 'comments': await post.comments}
但是，真正的问题 是当我有多个关系时：return 一个用户及其 post 及其评论。在那种情况下，我不得不以一种非常丑陋的方式（听起来不太好）将我的整个模型转换为 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',
]
main.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 在那些异步字段上自动等待（这样我就可以 return Post 实例）。 pydantic 怎么可能？
将 /posts/{id} 更改为 return post 及其 owner 没有注释实际上是使用这种方式工作（感谢@papple23j）：
return await Post.get_or_none(id=id).prefetch_related('owner')
但不适用于反向外键。 select_related('comments') 也没有帮助，它正在提高 AttributeError: can't set attribute.
您可以尝试使用 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__}
（以下文字使用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：写一个属性函数和return对应的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()
对不起，我太笨了
我想到的一个解决方案是使用 tortoise.contrib.pydantic 包：
PPost = pydantic_model_creator(Post)
# used as
return await PPost.from_tortoise_orm(await Post.get_or_none(id=1))
但是根据
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.
我使用的解决方案
Fetching foreign keys can be done with both async and sync interfaces.
Async fetch:
events = await tournament.events.all()Sync usage requires that you call fetch_related before the time, and then you can use common functions.
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))
Remember, a few hours of trial and error can save you several minutes of looking at the README.