文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2023-07-05 23:17

关注

本文小编为大家详细介绍“pydantic-resolve嵌套数据结构怎么生成LoaderDepend管理contextvars”,内容详细,步骤清晰,细节处理妥当,希望这篇“pydantic-resolve嵌套数据结构怎么生成LoaderDepend管理contextvars”文章能帮助大家解决疑惑,下面跟着小编的思路慢慢深入,一起来学习新知识吧。

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

pydantic-resolve

和GraphQL相比

结论:

GraphQL更适合 public API。

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

和 ORM 的 relationship相比

结论

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

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

LoaderDepend的用途 背景

如果你使用过dataloader, 不论是js还是python的,都会遇到一个问题,如何为单独的一个请求创建独立的dataloader?

以 python 的 strawberry 来举例子:

@strawberry.typeclass User:    id: strawberry.IDasync def load_users(keys) -> List[User]:    return [User(id=key) for key in keys]loader = DataLoader(load_fn=load_users)@strawberry.typeclass 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.typeclass User:    id: strawberry.IDasync 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.typeclass 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。

这个方法虽然好,但存在两个问题:

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

解决方法

我所期望的功能是:

其实这两件事情说的是同一个问题,就是如何把初始化的事情依赖反转到 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的过程,于是有一个思路:

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

下图中 1 会执行LoaderA 初始化,2,3则是读取缓存, 1.1 会执行LoaderB初始化,2.1,3.1 读取缓存

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

代码如下:

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”文章已经介绍完毕,想要掌握这篇文章的知识点还需要大家自己动手实践使用过才能领会,如果想了解更多相关内容的文章,欢迎关注编程网行业资讯频道。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯