pydantic-resolve嵌套數(shù)據(jù)結(jié)構(gòu)生成LoaderDepend管理contextvars
pydantic-resolve 解決嵌套數(shù)據(jù)結(jié)構(gòu)的生成和其他方案的比較
和GraphQL相比
- GraphQL的優(yōu)勢是 1.方便構(gòu)建嵌套結(jié)構(gòu),2.client可以方便生成查詢子集。非常適合構(gòu)建滿足靈活變化的 public API的場景.
- 但是很多實際業(yè)務在前端做的其實是照單全收,并沒有靈活選擇的需要。GraphQL帶來的便利更多體現(xiàn)在靈活地構(gòu)建嵌套結(jié)構(gòu)。
- GraphQL需要client端維護查詢語句,相較于通過
openapi.json
和工具自動生成client讓前后端無縫對接的做法,在前后端一體的架構(gòu)中維護這些查詢語句,屬于重復勞動。 - 為了滿足權(quán)限控制的需要,通過RESTful定義一個個API 會比全局一個Query,Mutation 控制起來更加清晰直接。
- Pydantic-resolve 恰好滿足了靈活構(gòu)建嵌套結(jié)構(gòu)的需求,它不需要像GraphQL一樣引入一系列概念和設(shè)置,它非常輕量級,沒有任何侵入,所有的功能通過簡單
resolve
一下就實現(xiàn)。 - Pydantic-resolve 在保持輕量級的同時,可以隱藏 Dataloader 的初始化邏輯,避免了GraphQL中在多處維護dataloader的麻煩。
- Pydantic-resolve 還提供了對 global
loader filters
的支持,在一些業(yè)務邏輯下可以簡化很多代碼。如果把Dataloader 的 keys 等價視為 relationship的 join on 條件的話, 那么loader_filters
就類似在別處的其他過濾條件。
結(jié)論:
GraphQL更適合 public API。
對前后端作為一個整體的項目,RESTful + Pydantic-resolve 才是快速靈活提供數(shù)據(jù)結(jié)構(gòu)的最佳方法。
和 ORM 的 relationship相比
- relationship 提供了ORM 級別的嵌套查詢實現(xiàn),但默認會使用lazy select的方法, 會導致很多的查詢次數(shù), 并且在異步使用的時候需要手動聲明例如
.option(subquery(Model.field))
之類的代碼 - relationship 的外鍵決定了,無法在關(guān)聯(lián)查詢的時候提供額外的過濾條件 (即便可以也是改動成本比較大的做法)
- relationship 最大的問題是使得 ORM Model 和 schema 產(chǎn)生了代碼耦合。在schema層想做的嵌套查詢,會把邏輯侵入到ORM Model層。
- Pydantic-resolve 則沒有這樣的問題,在 ORM 層不需要定義任何relationship,所有的join邏輯都通過 dataloader 批量查詢解決。 并且通過 global
loader_filters
參數(shù),可以提供額外的全局過濾條件。
結(jié)論
relationship 方案的靈活度低,不方便修改,默認的用法會產(chǎn)生外鍵約束。對迭代頻繁的項目不友好。
Pydantic-resolve 和 ORM 層完全解耦,可以通過靈活創(chuàng)建Dataloader 來滿足各種需要。
LoaderDepend的用途 背景
如果你使用過dataloader, 不論是js還是python的,都會遇到一個問題,如何為單獨的一個請求創(chuàng)建獨立的dataloader?
以 python 的 strawberry
來舉例子:
@strawberry.type class User: id: strawberry.ID async def load_users(keys) -> List[User]: return [User(id=key) for key in keys] loader = DataLoader(load_fn=load_users) @strawberry.type class Query: @strawberry.field async def get_user(self, id: strawberry.ID) -> User: return await loader.load(id) schema = strawberry.Schema(query=Query)
如果單獨實例化的話,會導致所有的請求都使用同一個dataloader, 由于loader本身是有緩存優(yōu)化機制的,所以即使內(nèi)容更新之后,依然會返回緩存的歷史數(shù)據(jù)。
因此 strawberry
的處理方式是:
@strawberry.type class User: id: strawberry.ID async def load_users(keys) -> List[User]: return [User(id=key) for key in keys] class MyGraphQL(GraphQL): async def get_context( self, request: Union[Request, WebSocket], response: Optional[Response] ) -> Any: return {"user_loader": DataLoader(load_fn=load_users)} @strawberry.type class Query: @strawberry.field async def get_user(self, info: Info, id: strawberry.ID) -> User: return await info.context["user_loader"].load(id) schema = strawberry.Schema(query=Query) app = MyGraphQL(schema)
開發(fā)者需要在get_context
中去初始化loader, 然后框架會負責在每次request的時候會執(zhí)行初始化。 這樣每個請求就會有獨立的loader, 解決了多次請求被緩存的問題。
其中的原理是:contextvars 在 await 的時候會做一次淺拷貝,所以外層的context可以被內(nèi)部讀到,因此手動在最外層(request的時候) 初始化一個引用類型(dict)之后,那么在 request 內(nèi)部自然就能獲取到引用類型內(nèi)的loader。
這個方法雖然好,但存在兩個問題:
- 需要手動去維護
get_context
, 每當新增了一個 DataLoader, 就需要去里面添加, 而且實際執(zhí)行.load
的地方也要從context 里面取loader。 - 存在初始化了loaders卻沒有被使用到的情況,比如整個Query 有 N 個loader,但是用戶的查詢實際只用到了1個,那么其他loader 的初始化就浪費了。而且作為公共區(qū)域東西多了之后代碼維護會不清晰。(重要)
而 graphene
就更加任性了,把loader 的活交給了 aiodataloader, 如果翻閱文檔的話,會發(fā)現(xiàn)處理的思路也是類似的,只是需要手動去維護創(chuàng)建過程。
解決方法
我所期望的功能是:
- 初始化按需執(zhí)行,比如我的整個schema 里面只存在 DataLoaderA, 那我希望只有DataLoaderA 被實例化
- 不希望在某個reqeust或者 middleware中干手動維護初始化。
其實這兩件事情說的是同一個問題,就是如何把初始化的事情依賴反轉(zhuǎn)到 resolve_field 方法中。
具體轉(zhuǎn)化為代碼:
class CommentSchema(BaseModel): id: int task_id: int content: str feedbacks: List[FeedbackSchema] = [] def resolve_feedbacks(self, loader=LoaderDepend(FeedbackLoader)): return loader.load(self.id) class TaskSchema(BaseModel): id: int name: str comments: List[CommentSchema] = [] def resolve_comments(self, loader=LoaderDepend(CommentLoader)): return loader.load(self.id)
就是說,我只要這樣申明好loader,其他的事情就一律不用操心。那么,這做得到么?
得益于pydantic-resolve
存在一個手動執(zhí)行resolve
的過程,于是有一個思路:
- contextvar 是淺拷貝,所以存的如果是引用類型,那么在最外層定義的dict,可以被所有內(nèi)層讀到??梢栽赗esolver初始化的時候定義。
- 假如
tasks: list[TaskSchema]
有n個,我希望在第一次遇到的時候把loader 初始化并緩存,后續(xù)其他都使用緩存的loader。 - LoaderDepend 里面存放的是 DataLoader類,做為default 參數(shù)傳入resolve_field 方法
- 執(zhí)行resolve_field之前,利用inspect.signature 分析 default 參數(shù),執(zhí)行初始化和緩存的邏輯。
總體就是一個lazy的路子,到實際執(zhí)行的時候去處理初始化流程。
下圖中 1 會執(zhí)行LoaderA 初始化,2,3則是讀取緩存, 1.1 會執(zhí)行LoaderB初始化,2.1,3.1 讀取緩存
代碼如下:
class Resolver: def __init__(self): self.ctx = contextvars.ContextVar('pydantic_resolve_internal_context', default={}) def exec_method(self, method): signature = inspect.signature(method) params = {} for k, v in signature.parameters.items(): if isinstance(v.default, Depends): cache_key = str(v.default.dependency.__name__) cache = self.ctx.get() hit = cache.get(cache_key, None) if hit: instance = hit else: instance = v.default.dependency() cache[cache_key] = instance self.ctx.set(cache) params[k] = instance return method(**params)
遺留問題 (已經(jīng)解決)
有些DataLoader的實現(xiàn)可能需要一個外部的查詢條件, 比如查詢用戶的absense信息的時候,除了user_key 之外,還需要額外提供其他全局filter 比如sprint_id)。 這種全局變量從load參數(shù)走會顯得非常啰嗦。
這種時候就依然需要借助contextvars 在外部設(shè)置變量。 以一段項目代碼為例:
async def get_team_users_load(team_id: int, sprint_id: Optional[int], session: AsyncSession): ctx.team_id_context.set(team_id) # set global filter ctx.sprint_id_context.set(sprint_id) # set global filter res = await session.execute(select(User) .join(UserTeam, UserTeam.user_id == User.id) .filter(UserTeam.team_id == team_id)) db_users = res.scalars() users = [schema.UserLoadUser(id=u.id, employee_id=u.employee_id, name=u.name) for u in db_users] results = await Resolver().resolve(users) # resolve return results
class AbsenseLoader(DataLoader): async def batch_load_fn(self, user_keys): async with async_session() as session, session.begin(): sprint_id = ctx.sprint_id_context.get() # read global filter sprint_stmt = Sprint.status == SprintStatusEnum.ongoing if not sprint_id else Sprint.id == sprint_id res = await session.execute(select(SprintAbsence) .join(Sprint, Sprint.id == SprintAbsence.sprint_id) .join(User, User.id == SprintAbsence.user_id) .filter(sprint_stmt) .filter(SprintAbsence.user_id.in_(user_keys))) rows = res.scalars().all() dct = {} for row in rows: dct[row.user_id] = row.hours return [dct.get(k, 0) for k in user_keys]
期望的設(shè)置方式為:
loader_filters = { AbsenseLoader: {'sprint_id': 10}, OtherLoader: {field: 'value_x'} } results = await Resolver(loader_filters=loader_filters).resolve(users)
如果需要filter但是卻沒有設(shè)置, 該情況下要拋異常
以上就是pydantic-resolve嵌套數(shù)據(jù)結(jié)構(gòu)生成LoaderDepend管理contextvars的詳細內(nèi)容,更多關(guān)于LoaderDepend管理contextvars的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
基于python,Matplotlib繪制函數(shù)的等高線與三維圖像
這篇文章主要介紹了基于python,Matplotlib繪制函數(shù)的等高線與三維圖像,函數(shù)的等高線及其三維圖像的可視化方法,下面一起來學習具體內(nèi)容吧,需要的小伙伴可以參考一下2022-01-01Django配置Bootstrap, js實現(xiàn)過程詳解
這篇文章主要介紹了Django配置Bootstrap, js實現(xiàn)過程詳解,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-10-10OpenCV圖像縮放resize各種插值方式的比較實現(xiàn)
OpenCV提供了resize函數(shù)來改變圖像的大小,本文主要介紹了OpenCV圖像縮放resize各種插值方式的比較實現(xiàn),分享給大家,感興趣的可以了解一下2021-06-06