pydantic?resolve解決嵌套數(shù)據(jù)結(jié)構(gòu)生成痛點(diǎn)分析
案例
以論壇為例，有個(gè)接口返回帖子(posts)信息，然后呢，來(lái)了新需求，說(shuō)需要顯示帖子的 author 信息。
此時(shí)會(huì)有兩種選擇：
在 posts 的 query 中 join 查詢 author 信息，在返回 post 中添加諸如 author_id, author_name 之類的字段。
{'post': 'v2ex', 'author_name': 'tangkikodo'}
根據(jù) posts 的 ids ， 單獨(dú)查詢 author 列表，然后把 author 對(duì)象循環(huán)添加到 post 對(duì)象中。
{'post':'v2ex', 'author': {'name': 'tangkikod'}}
方法 1 中，需要去修改 query, 還需要修改post的schema. 如果未來(lái)要加新字段，例如用戶頭像的話，會(huì)需要修改兩處。
方法 2 需要手動(dòng)做一次拼接。之后增減字段都是在 author 對(duì)象的范圍內(nèi)修改。
所以相對(duì)來(lái)說(shuō), 方法 2 在未來(lái)的可維護(hù)性會(huì)比較好。用嵌套對(duì)象的方式可以更好的擴(kuò)展和維護(hù)。
方法2 的返回結(jié)構(gòu)
[
{
"id": 1,
"post": "v2ex",
"author": {
"name": "tangkikodo",
"id": 1
}
},
{
"id": 2,
"post": "v3ex",
"author": {
"name": "tangkikodo2",
"id": 1
}
}
]
然而需求總是會(huì)變化，突然來(lái)了一個(gè)新的且奇怪的需求，要在 author 信息中添加數(shù)據(jù)，顯示他最近瀏覽過(guò)的帖子。返回體變成了：
[
{
"id": 1,
"post": "v2ex",
"author": {
"name": "tangkikodo",
"recent_views": [
{
"id": 2,
"post": "v3ex"
},
{
"id": 3,
"post": "v4ex"
}
]
}
}
]
那這個(gè)時(shí)候該怎么弄呢？血壓是不是有點(diǎn)上來(lái)了。
根據(jù)之前的方法 2, 通常的操作是在獲取到authors信息后, 關(guān)聯(lián)查找author的recent_posts, 拼接回authors, 再將 authors 拼接回posts。 流程類似層層查找再層層回拼。 偽代碼類似：
# posts query posts = query_all_posts() # authors query authors_ids = fetch_unique_author_id(posts) authors = query_author(author_ids) recent_view_posts = fetch_recent_review_posts(author_ids) # 新需求 recent_view_maps = calc_view_mapping(recent_view_posts) # 新需求 # authors attach authors = [attach_posts(a, recent_view_maps) for a in authors] author_map = calc_author_mapping(authors) # posts attach posts = [attach_author(p, author_map) for p in posts]
莫名的會(huì)聯(lián)想到callback hell, 添加新的層級(jí)都會(huì)在代碼中間部分切入。
反正想想就挺麻煩的對(duì)吧。要是哪天再嵌套一層呢? 代碼改起來(lái)有點(diǎn)費(fèi)勁, 如果你此時(shí)血壓有點(diǎn)高，那請(qǐng)繼續(xù)往下看。
那，有別的辦法么？ 這里有個(gè)小輪子也許能幫忙。
解決方法
祭出一個(gè)小輪子： allmonday/pydantic-resolve
以剛才的例子，要做的事情抽象成兩部分:
- 定義 dataloader ，負(fù)責(zé)查詢和group數(shù)據(jù)。前半部分是從數(shù)據(jù)庫(kù)查詢，后半部分是將數(shù)據(jù)轉(zhuǎn)成 pydantic 對(duì)象后返回。 偽代碼，看個(gè)大概意思就好。
class AuthorLoader(DataLoader):
async def batch_load_fn(self, author_ids):
async with async_session() as session:
# query authors
res = await session.execute(select(Author).where(Author.id.in_(author_ids)))
rows = res.scalars().all()
# transform into pydantic object
dct = defaultdict(dict)
for row in rows:
dct[row.author_id] = AuthorSchema.from_orm(row)
# order by author_id
return [dct.get(k, None) for k in author_ids]
class RecentViewPostLoader(DataLoader):
async def batch_load_fn(self, view_ids):
async with async_session() as session:
res = await session.execute(select(Post, PostVisit.visitor_id) # join 瀏覽中間表
.join(PostVist, PostVisit.post_id == Post.id)
.where(PostVisit.user_id.in_(view_ids)
.where(PostVisit.created_at < some_timestamp)))
rows = res.scalars().all()
dct = defaultdict(list)
for row in rows:
dct[row.visitor_id].append(PostSchema.from_orm(row)) # group 到 visitor
return [dct.get(k, []) for k in view_ids]
- 定義 schema, 并且注入依賴的 DataLoaders, LoaderDepend 會(huì)管理好loader 的異步上下文緩存。
class RecentPostSchema(BaseModel):
id: int
name: str
class Config:
orm_mode = True
class AuthorSchema(BaseModel):
id: int
name: str
img_url: str
recent_views: Tuple[RecentPostSchema, ...] = tuple()
def resolve_recent_views(self, loader=LoaderDepend(RecentViewPostLoader)):
return loader.load(self.id)
class Config:
orm_mode = True
class PostSchema(BaseModel):
id: int
author_id: int
name: str
author: Optional[AuthorSchema] = None
def resolve_author(self, loader=LoaderDepend(AuthorLoader)):
return loader.load(self.author_id)
class Config:
orm_mode = True
然后呢？
然后就沒(méi)有了，接下來(lái)只要做個(gè) post 的查詢, 再簡(jiǎn)單地...resolve 一下，任務(wù)就完成了。
posts = (await session.execute(select(Post))).scalars().all() posts = [PostSchema.from_orm(p) for p in posts] results = await Resolver().resolve(posts)
在拆分了 loader 和 schema 之后，對(duì)數(shù)據(jù)地任意操作都很簡(jiǎn)單，添加任意新的schema 都不會(huì)破壞原有的代碼。
完整的案例可以查看 6_sqlalchemy_loaderdepend_global_filter.py
如果之前使用過(guò)aiodataloader 的話會(huì)知道，開(kāi)發(fā)需要手動(dòng)維護(hù)loader在每個(gè)request 中的初始化過(guò)程 ， 但在 pydantic-resolve 中你完全不用操心異步上下文的創(chuàng)建，不用維護(hù)DataLoader的實(shí)例化, 一切都在pydantic-resolve的管理之中。
就完事了。如果必須說(shuō)有啥缺點(diǎn)的話。。必須用 async await 可能算一個(gè)。
該項(xiàng)目已經(jīng)在我司的生產(chǎn)環(huán)境中使用，并且保持了100%的測(cè)試覆蓋率。 歡迎大家嘗鮮體驗(yàn)，如果遇到問(wèn)題歡迎發(fā)issue，我會(huì)盡快修復(fù)。
以上就是pydantic resolve解決嵌套數(shù)據(jù)結(jié)構(gòu)生成痛點(diǎn)分析的詳細(xì)內(nèi)容，更多關(guān)于pydantic resolve嵌套數(shù)據(jù)結(jié)構(gòu)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
Python?GUI實(shí)現(xiàn)PDF轉(zhuǎn)Word功能
這篇文章主要介紹了如何使用?wxPython?創(chuàng)建一個(gè)簡(jiǎn)單的圖形用戶界面（GUI）應(yīng)用程序,結(jié)合?pdf2docx?庫(kù),實(shí)現(xiàn)將?PDF?轉(zhuǎn)換為?Word?文檔的功能,需要的可以參考下2024-12-12
python實(shí)現(xiàn)從網(wǎng)絡(luò)下載文件并獲得文件大小及類型的方法
這篇文章主要介紹了python實(shí)現(xiàn)從網(wǎng)絡(luò)下載文件并獲得文件大小及類型的方法,涉及Python操作網(wǎng)絡(luò)文件的相關(guān)技巧,需要的朋友可以參考下2015-04-04
選擇了腳本語(yǔ)言就要忍受其速度，這句話在某種程度上說(shuō)明了 python 作為腳本的一個(gè)不足之處，那就是執(zhí)行效率和性能不夠理想，特別是在 performance 較差的機(jī)器上，因此有必要進(jìn)行一定的代碼優(yōu)化來(lái)提高程序的執(zhí)行效率2012-08-08
這篇文章主要介紹了Python OS模塊常用函數(shù)說(shuō)明,本文列出了一些在os模塊中比較有用的部分函數(shù),它們中的大多數(shù)都簡(jiǎn)單明了,需要的朋友可以參考下2015-05-05
pytorch多GPU并行運(yùn)算的實(shí)現(xiàn)
這篇文章主要介紹了pytorch多GPU并行運(yùn)算的實(shí)現(xiàn)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09
關(guān)于pyinstaller?打包多個(gè)py文件的問(wèn)題
這篇文章主要介紹了pyinstaller?打包多個(gè)py文件及遇到的問(wèn)題，本文通過(guò)實(shí)例代碼給大家詳細(xì)介紹，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2022-10-10
Python編程OpenCV和Numpy圖像處理庫(kù)實(shí)現(xiàn)圖片去水印
這篇文章主要介紹了Python編程中如何實(shí)現(xiàn)圖片去水印本文采用了OpenCV和Numpy的圖像處理的方法來(lái)實(shí)現(xiàn)，文中附含詳細(xì)示例代碼，有需要的朋友可以借鑒參考下2021-09-09
本篇文章主要是介紹了Python的Lambda函數(shù)與排序，簡(jiǎn)單的介紹了Lambda函數(shù)的用法和排序，有需要的朋友可以了解一下。2016-10-10
python神經(jīng)網(wǎng)絡(luò)facenet人臉檢測(cè)及keras實(shí)現(xiàn)
這篇文章主要為大家介紹了python神經(jīng)網(wǎng)絡(luò)facenet人臉檢測(cè)及keras實(shí)現(xiàn)，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
這篇文章主要介紹了Python判斷列表是否已排序的各種方法及其性能分析的相關(guān)資料,需要的朋友可以參考下2016-06-06