pydantic-resolve嵌套数据结构生成LoaderDepend管理contextvars

更新时间：2023年04月07日 11:27:57 作者：allmonday

这篇文章主要为大家介绍了pydantic-resolve解决嵌套数据结构生成LoaderDepend管理contextvars的使用示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪<BR>

pydantic-resolve 解决嵌套数据结构的生成和其他方案的比较

pydantic-resolve

和GraphQL相比

GraphQL的优势是 1.方便构建嵌套结构，2.client可以方便生成查询子集。非常适合构建满足灵活变化的 public API的场景.
但是很多实际业务在前端做的其实是照单全收，并没有灵活选择的需要。GraphQL带来的便利更多体现在灵活地构建嵌套结构。
GraphQL需要client端维护查询语句，相较于通过openapi.json和工具自动生成client让前后端无缝对接的做法，在前后端一体的架构中维护这些查询语句，属于重复劳动。
为了满足权限控制的需要，通过RESTful定义一个个API 会比全局一个Query，Mutation 控制起来更加清晰直接。
Pydantic-resolve 恰好满足了灵活构建嵌套结构的需求，它不需要像GraphQL一样引入一系列概念和设置，它非常轻量级，没有任何侵入，所有的功能通过简单resolve一下就实现。
Pydantic-resolve 在保持轻量级的同时，可以隐藏 Dataloader 的初始化逻辑，避免了GraphQL中在多处维护dataloader的麻烦。
Pydantic-resolve 还提供了对 global loader filters 的支持，在一些业务逻辑下可以简化很多代码。如果把Dataloader 的 keys 等价视为 relationship的 join on 条件的话，那么 loader_filters 就类似在别处的其他过滤条件。

结论：

GraphQL更适合 public API。

对前后端作为一个整体的项目，RESTful + Pydantic-resolve 才是快速灵活提供数据结构的最佳方法。

和 ORM 的 relationship相比

relationship 提供了ORM 级别的嵌套查询实现，但默认会使用lazy select的方法，会导致很多的查询次数，并且在异步使用的时候需要手动声明例如 .option(subquery(Model.field)) 之类的代码
relationship 的外键决定了，无法在关联查询的时候提供额外的过滤条件（即便可以也是改动成本比较大的做法）
relationship 最大的问题是使得 ORM Model 和 schema 产生了代码耦合。在schema层想做的嵌套查询，会把逻辑侵入到ORM Model层。
Pydantic-resolve 则没有这样的问题，在 ORM 层不需要定义任何relationship，所有的join逻辑都通过 dataloader 批量查询解决。并且通过 global loader_filters 参数，可以提供额外的全局过滤条件。

结论

relationship 方案的灵活度低，不方便修改，默认的用法会产生外键约束。对迭代频繁的项目不友好。

Pydantic-resolve 和 ORM 层完全解耦，可以通过灵活创建Dataloader 来满足各种需要。

LoaderDepend的用途背景

如果你使用过dataloader, 不论是js还是python的，都会遇到一个问题，如何为单独的一个请求创建独立的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本身是有缓存优化机制的，所以即使内容更新之后，依然会返回缓存的历史数据。

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

开发者需要在get_context中去初始化loader, 然后框架会负责在每次request的时候会执行初始化。这样每个请求就会有独立的loader, 解决了多次请求被缓存的问题。

其中的原理是：contextvars 在 await 的时候会做一次浅拷贝，所以外层的context可以被内部读到，因此手动在最外层（request的时候) 初始化一个引用类型(dict)之后，那么在 request 内部自然就能获取到引用类型内的loader。

这个方法虽然好，但存在两个问题：

需要手动去维护 get_context, 每当新增了一个 DataLoader, 就需要去里面添加, 而且实际执行 .load 的地方也要从context 里面取loader。
存在初始化了loaders却没有被使用到的情况，比如整个Query 有 N 个loader，但是用户的查询实际只用到了1个，那么其他loader 的初始化就浪费了。而且作为公共区域东西多了之后代码维护会不清晰。(重要)

而 graphene 就更加任性了，把loader 的活交给了 aiodataloader, 如果翻阅文档的话，会发现处理的思路也是类似的，只是需要手动去维护创建过程。

解决方法

我所期望的功能是：

初始化按需执行，比如我的整个schema 里面只存在 DataLoaderA, 那我希望只有DataLoaderA 被实例化
不希望在某个reqeust或者 middleware中干手动维护初始化。

其实这两件事情说的是同一个问题，就是如何把初始化的事情依赖反转到 resolve_field 方法中。

具体转化为代码：

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 存在一个手动执行resolve的过程，于是有一个思路：

contextvar 是浅拷贝，所以存的如果是引用类型，那么在最外层定义的dict，可以被所有内层读到。可以在Resolver初始化的时候定义。
假如 tasks: list[TaskSchema] 有n个，我希望在第一次遇到的时候把loader 初始化并缓存，后续其他都使用缓存的loader。
LoaderDepend 里面存放的是 DataLoader类，做为default 参数传入resolve_field 方法
执行resolve_field之前，利用inspect.signature 分析 default 参数，执行初始化和缓存的逻辑。

总体就是一个lazy的路子，到实际执行的时候去处理初始化流程。

下图中 1 会执行LoaderA 初始化，2，3则是读取缓存, 1.1 会执行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)

遗留问题 (已经解决）

有些DataLoader的实现可能需要一个外部的查询条件，比如查询用户的absense信息的时候，除了user_key 之外，还需要额外提供其他全局filter 比如sprint_id)。这种全局变量从load参数走会显得非常啰嗦。

这种时候就依然需要借助contextvars 在外部设置变量。以一段项目代码为例：

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]

期望的设置方式为：

loader_filters = {
    AbsenseLoader: {'sprint_id': 10}, 
    OtherLoader: {field: 'value_x'}
}
results = await Resolver(loader_filters=loader_filters).resolve(users)

如果需要filter但是却没有设置, 该情况下要抛异常

以上就是pydantic-resolve嵌套数据结构生成LoaderDepend管理contextvars的详细内容，更多关于LoaderDepend管理contextvars的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

用python爬取历史天气数据的方法示例
这篇文章主要介绍了用python爬取历史天气数据的方法示例，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-12-12
Python实现自动化发送邮件
大家好，本篇文章主要讲的是Python实现自动化发送邮件，感兴趣的同学赶快来看一看吧，对你有帮助的话记得收藏一下
2022-01-01
Python爬虫爬取爱奇艺电影片库首页的实例代码
这篇文章主要介绍了Python爬虫爬取爱奇艺电影片库首页的实例代码,代码简单易懂，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-05-05
Selenium+Python自动化脚本环境搭建的全过程
说到自动化测试,就不得不提大名鼎鼎的Selenium,Selenium 是如今最常用的自动化测试工具之一,支持快速开发自动化测试框架,且支持在多种浏览器上执行测试,下面这篇文章主要给大家介绍了关于Selenium+Python自动化脚本环境搭建的相关资料,需要的朋友可以参考下
2021-09-09
浅谈Keras中shuffle和validation_split的顺序
这篇文章主要介绍了浅谈Keras中shuffle和validation_split的顺序，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-06-06
终端命令查看TensorFlow版本号及路径的方法
今天小编就为大家分享一篇终端命令查看TensorFlow版本号及路径的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-06-06
在win64上使用bypy进行百度网盘文件上传功能
这篇文章主要介绍了在win64上使用bypy进行百度网盘文件上传功能，本文给大家介绍的非常详细，具有一定的参考借鉴价值,需要的朋友可以参考下
2020-01-01
Python全栈之运算符详解
这篇文章主要为大家介绍了Python运算符，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助
2021-12-12
Python range、enumerate和zip函数用法详解
这篇文章主要介绍了Python range、enumerate和zip函数用法详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-09-09
Python中不同类之间调用方法的四种方式小结
类是一种面向对象的编程范式,它允许我们将数据和功能封装在一个实体中,本文主要介绍了Python中不同类之间调用方法的四种方式小结,具有一定的参考价值,感兴趣的可以了解一下
2024-02-02