欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

pydantic-resolve嵌套數(shù)據(jù)結(jié)構(gòu)生成LoaderDepend管理contextvars

 更新時(shí)間:2023年04月07日 11:27:57   作者:allmonday  
這篇文章主要為大家介紹了pydantic-resolve解決嵌套數(shù)據(jù)結(jié)構(gòu)生成LoaderDepend管理contextvars的使用示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪<BR>

pydantic-resolve 解決嵌套數(shù)據(jù)結(jié)構(gòu)的生成和其他方案的比較

pydantic-resolve

和GraphQL相比

  • GraphQL的優(yōu)勢是 1.方便構(gòu)建嵌套結(jié)構(gòu),2.client可以方便生成查詢子集。非常適合構(gòu)建滿足靈活變化的 public API的場景.
  • 但是很多實(shí)際業(yè)務(wù)在前端做的其實(shí)是照單全收,并沒有靈活選擇的需要。GraphQL帶來的便利更多體現(xiàn)在靈活地構(gòu)建嵌套結(jié)構(gòu)。
  • GraphQL需要client端維護(hù)查詢語句,相較于通過openapi.json和工具自動(dòng)生成client讓前后端無縫對(duì)接的做法,在前后端一體的架構(gòu)中維護(hù)這些查詢語句,屬于重復(fù)勞動(dòng)。
  • 為了滿足權(quán)限控制的需要,通過RESTful定義一個(gè)個(gè)API 會(huì)比全局一個(gè)Query,Mutation 控制起來更加清晰直接。
  • Pydantic-resolve 恰好滿足了靈活構(gòu)建嵌套結(jié)構(gòu)的需求,它不需要像GraphQL一樣引入一系列概念和設(shè)置,它非常輕量級(jí),沒有任何侵入,所有的功能通過簡單resolve一下就實(shí)現(xiàn)。
  • Pydantic-resolve 在保持輕量級(jí)的同時(shí),可以隱藏 Dataloader 的初始化邏輯,避免了GraphQL中在多處維護(hù)dataloader的麻煩。
  • Pydantic-resolve 還提供了對(duì) global loader filters 的支持,在一些業(yè)務(wù)邏輯下可以簡化很多代碼。如果把Dataloader 的 keys 等價(jià)視為 relationship的 join on 條件的話, 那么 loader_filters 就類似在別處的其他過濾條件。

結(jié)論:

GraphQL更適合 public API。

對(duì)前后端作為一個(gè)整體的項(xiàng)目,RESTful + Pydantic-resolve 才是快速靈活提供數(shù)據(jù)結(jié)構(gòu)的最佳方法。

和 ORM 的 relationship相比

  • relationship 提供了ORM 級(jí)別的嵌套查詢實(shí)現(xiàn),但默認(rèn)會(huì)使用lazy select的方法, 會(huì)導(dǎo)致很多的查詢次數(shù), 并且在異步使用的時(shí)候需要手動(dòng)聲明例如 .option(subquery(Model.field)) 之類的代碼
  • relationship 的外鍵決定了,無法在關(guān)聯(lián)查詢的時(shí)候提供額外的過濾條件 (即便可以也是改動(dòng)成本比較大的做法)
  • relationship 最大的問題是使得 ORM Model 和 schema 產(chǎn)生了代碼耦合。在schema層想做的嵌套查詢,會(huì)把邏輯侵入到ORM Model層。
  • Pydantic-resolve 則沒有這樣的問題,在 ORM 層不需要定義任何relationship,所有的join邏輯都通過 dataloader 批量查詢解決。 并且通過 global loader_filters 參數(shù),可以提供額外的全局過濾條件。

結(jié)論

relationship 方案的靈活度低,不方便修改,默認(rèn)的用法會(huì)產(chǎn)生外鍵約束。對(duì)迭代頻繁的項(xiàng)目不友好。

Pydantic-resolve 和 ORM 層完全解耦,可以通過靈活創(chuàng)建Dataloader 來滿足各種需要。

LoaderDepend的用途 背景

如果你使用過dataloader, 不論是js還是python的,都會(huì)遇到一個(gè)問題,如何為單獨(dú)的一個(gè)請(qǐng)求創(chuàng)建獨(dú)立的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)

如果單獨(dú)實(shí)例化的話,會(huì)導(dǎo)致所有的請(qǐng)求都使用同一個(gè)dataloader, 由于loader本身是有緩存優(yōu)化機(jī)制的,所以即使內(nèi)容更新之后,依然會(huì)返回緩存的歷史數(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, 然后框架會(huì)負(fù)責(zé)在每次request的時(shí)候會(huì)執(zhí)行初始化。 這樣每個(gè)請(qǐng)求就會(huì)有獨(dú)立的loader, 解決了多次請(qǐng)求被緩存的問題。

其中的原理是:contextvars 在 await 的時(shí)候會(huì)做一次淺拷貝,所以外層的context可以被內(nèi)部讀到,因此手動(dòng)在最外層(request的時(shí)候) 初始化一個(gè)引用類型(dict)之后,那么在 request 內(nèi)部自然就能獲取到引用類型內(nèi)的loader。

這個(gè)方法雖然好,但存在兩個(gè)問題:

  • 需要手動(dòng)去維護(hù) get_context, 每當(dāng)新增了一個(gè) DataLoader, 就需要去里面添加, 而且實(shí)際執(zhí)行 .load 的地方也要從context 里面取loader。
  • 存在初始化了loaders卻沒有被使用到的情況,比如整個(gè)Query 有 N 個(gè)loader,但是用戶的查詢實(shí)際只用到了1個(gè),那么其他loader 的初始化就浪費(fèi)了。而且作為公共區(qū)域東西多了之后代碼維護(hù)會(huì)不清晰。(重要)

graphene 就更加任性了,把loader 的活交給了 aiodataloader, 如果翻閱文檔的話,會(huì)發(fā)現(xiàn)處理的思路也是類似的,只是需要手動(dòng)去維護(hù)創(chuàng)建過程。

解決方法

我所期望的功能是:

  • 初始化按需執(zhí)行,比如我的整個(gè)schema 里面只存在 DataLoaderA, 那我希望只有DataLoaderA 被實(shí)例化
  • 不希望在某個(gè)reqeust或者 middleware中干手動(dòng)維護(hù)初始化。

其實(shí)這兩件事情說的是同一個(gè)問題,就是如何把初始化的事情依賴反轉(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 存在一個(gè)手動(dòng)執(zhí)行resolve的過程,于是有一個(gè)思路:

  • contextvar 是淺拷貝,所以存的如果是引用類型,那么在最外層定義的dict,可以被所有內(nèi)層讀到。可以在Resolver初始化的時(shí)候定義。
  • 假如 tasks: list[TaskSchema] 有n個(gè),我希望在第一次遇到的時(shí)候把loader 初始化并緩存,后續(xù)其他都使用緩存的loader。
  • LoaderDepend 里面存放的是 DataLoader類,做為default 參數(shù)傳入resolve_field 方法
  • 執(zhí)行resolve_field之前,利用inspect.signature 分析 default 參數(shù),執(zhí)行初始化和緩存的邏輯。

總體就是一個(gè)lazy的路子,到實(shí)際執(zhí)行的時(shí)候去處理初始化流程。

下圖中 1 會(huì)執(zhí)行LoaderA 初始化,2,3則是讀取緩存, 1.1 會(huì)執(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的實(shí)現(xiàn)可能需要一個(gè)外部的查詢條件, 比如查詢用戶的absense信息的時(shí)候,除了user_key 之外,還需要額外提供其他全局filter 比如sprint_id)。 這種全局變量從load參數(shù)走會(huì)顯得非常啰嗦。

這種時(shí)候就依然需要借助contextvars 在外部設(shè)置變量。 以一段項(xiàng)目代碼為例:

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的詳細(xì)內(nèi)容,更多關(guān)于LoaderDepend管理contextvars的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • 基于python,Matplotlib繪制函數(shù)的等高線與三維圖像

    基于python,Matplotlib繪制函數(shù)的等高線與三維圖像

    這篇文章主要介紹了基于python,Matplotlib繪制函數(shù)的等高線與三維圖像,函數(shù)的等高線及其三維圖像的可視化方法,下面一起來學(xué)習(xí)具體內(nèi)容吧,需要的小伙伴可以參考一下
    2022-01-01
  • Django配置Bootstrap, js實(shí)現(xiàn)過程詳解

    Django配置Bootstrap, js實(shí)現(xiàn)過程詳解

    這篇文章主要介紹了Django配置Bootstrap, js實(shí)現(xiàn)過程詳解,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下
    2020-10-10
  • python實(shí)現(xiàn)udp傳輸圖片功能

    python實(shí)現(xiàn)udp傳輸圖片功能

    這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)udp傳輸圖片功能,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下
    2020-03-03
  • Python中的引用和拷貝實(shí)例解析

    Python中的引用和拷貝實(shí)例解析

    這篇文章主要介紹了python中的引用和拷貝實(shí)例解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下
    2019-11-11
  • Python?matplotlib底層原理解析

    Python?matplotlib底層原理解析

    這篇文章主要介紹了Python?matplotlib底層原理,下面文章圍繞Python?matplotlib底層原理的相關(guān)資料展開詳細(xì)內(nèi)容,具有一定的參考價(jià)值,需要的朋友可以參考下
    2021-12-12
  • python字典值排序并取出前n個(gè)key值的方法

    python字典值排序并取出前n個(gè)key值的方法

    今天小編就為大家分享一篇python字典值排序并取出前n個(gè)key值的方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧
    2018-10-10
  • python中的循環(huán)語法使用指南

    python中的循環(huán)語法使用指南

    這篇文章主要給大家介紹了關(guān)于python中循環(huán)語法使用的相關(guān)資料, 循環(huán)語句是Python中的一種基本語句,用于重復(fù)執(zhí)行一段代碼。在Python中,循環(huán)語句分為for和while兩種,需要的朋友可以參考下
    2023-08-08
  • OpenCV圖像縮放resize各種插值方式的比較實(shí)現(xiàn)

    OpenCV圖像縮放resize各種插值方式的比較實(shí)現(xiàn)

    OpenCV提供了resize函數(shù)來改變圖像的大小,本文主要介紹了OpenCV圖像縮放resize各種插值方式的比較實(shí)現(xiàn),分享給大家,感興趣的可以了解一下
    2021-06-06
  • Python collections模塊的使用方法

    Python collections模塊的使用方法

    這篇文章主要介紹了Python collections模塊的使用方法,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2020-10-10
  • Pyinstaller打包Pytorch框架所遇到的問題

    Pyinstaller打包Pytorch框架所遇到的問題

    Pytorch在python界用得比較多,打包容易失敗,本文主要介紹了Pyinstaller打包Pytorch框架所遇到的問題,文中介紹的十分詳盡,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下
    2022-03-03

最新評(píng)論