Graphene Graphql - 如何链接突变

Graphene Graphql - how to chain mutations

我碰巧向 Graphql API (Python3 + Graphene) 发送了 2 个单独的请求,以便:

  1. 创建对象
  2. 更新另一个对象,使其与创建的对象相关联。

我感觉到这可能不在 Graphql 的 "spirit" 中,所以我搜索并阅读了有关 nested migrations. Unforutnately, I also found that it was bad practice 的内容,因为嵌套迁移不是顺序的,它可能会导致客户难以调试问题,因为竞争条件。

我正在尝试使用顺序根突变来实现考虑嵌套迁移的用例。请允许我向您展示我想象的一个用例和一个简单的解决方案(但可能不是好的做法)。很抱歉这么久 post 来了。

假设我有用户和组实体,我希望从客户端表单更新组,不仅能够添加用户,还能创建要添加到组中的用户,如果用户不存在。用户有名为 uid (user id) 和 groups gid (groupd id) 的 id,只是为了突出区别。所以使用根突变,我想象做这样的查询:

mutation {
    createUser(uid: "b53a20f1b81b439", username: "new user", password: "secret"){
        uid
        username
    }

    updateGroup(gid: "group id", userIds: ["b53a20f1b81b439", ...]){
        gid
        name
    }
}

您注意到我在 createUser 突变的输入中提供了用户 ID。我的问题是要进行 updateGroup 突变,我需要新创建用户的 ID。我不知道如何在解析 updateGroup 的 mutate 方法中的石墨烯中获取它,所以我想象在加载客户端表单数据时从 API 查询 UUID。所以在发送上面的突变之前,在我的客户端初始加载时,我会做类似的事情:

query {
    uuid

    group (gid: "group id") {
        gid
        name
    }
}

然后我将在突变请求中使用来自此查询响应的 uuid(值将是 b53a20f1b81b439,如上面的第一个 scriptlet 中所示)。

你觉得这个过程怎么样?有更好的方法吗? Python uuid.uuid4 实施这个安全吗?

提前致谢。

----- 编辑

根据评论中的讨论,我应该提到上面的用例仅用于说明。事实上,一个用户实体可能有一个内在的唯一键(电子邮件、用户名),其他实体也可能有(图书的 ISBN...)。我正在寻找一般情况下的解决方案,包括可能不会展示此类自然唯一键的实体。

最初问题下的评论中有很多建议。我会在这个提案的最后回到一些。

我一直在思考这个问题,而且它似乎是开发人员中反复出现的问题。我得出的结论是,我们可能会在想要编辑图形的方式中遗漏一些东西,即边缘操作。我想我们尝试用节点操作来做边缘操作。为了说明这一点,使用点 (Graphviz) 等语言创建的图形可能如下所示:

digraph D {

  /* Nodes */
  A 
  B
  C

  /* Edges */

  A -> B
  A -> C
  A -> D

}

按照这种模式,问题中的 graphql 突变可能如下所示:

mutation {

    # Nodes

    n1: createUser(username: "new user", password: "secret"){
        uid
        username
    }

    n2: updateGroup(gid: "group id"){
        gid
        name
    }

    # Edges

    addUserToGroup(user: "n1", group: "n2"){
        status
    }
}

"edge operation"addUserToGroup的输入将是突变查询中先前节点的别名。

这也将允许通过权限检查来修饰边缘操作(创建关系的权限可能与每个对象的权限不同)。

我们肯定已经可以解决这样的查询了。不太确定的是,后端框架,尤其是 Graphene-python,是否提供允许实施 addUserToGroup 的机制(在解析上下文中产生先前的突变结果)。我正在考虑在石墨烯上下文中注入先前结果的 dict。如果成功,我会尝试用技术细节来完成答案。

也许已经有办法实现这样的目标了,我也会寻找它并在找到时完成答案。

如果事实证明上述模式不可能或发现不好的做法,我想我会坚持 2 个单独的突变。


编辑 1:共享结果

我测试了一种解决上述查询的方法,使用 Graphene-python middleware and a base mutation class to handle sharing the results. I created a one-file python program available on Github to test this. Or play with it on Repl.

中间件非常简单,将字典作为 kwarg 参数添加到解析器:

class ShareResultMiddleware:

    shared_results = {}

    def resolve(self, next, root, info, **args):
        return next(root, info, shared_results=self.shared_results, **args)

基础class也很简单,管理字典中结果的插入:

class SharedResultMutation(graphene.Mutation):

    @classmethod
    def mutate(cls, root: None, info: graphene.ResolveInfo, shared_results: dict, *args, **kwargs):
        result = cls.mutate_and_share_result(root, info, *args, **kwargs)
        if root is None:
            node = info.path[0]
            shared_results[node] = result
        return result

    @staticmethod
    def mutate_and_share_result(*_, **__):
        return SharedResultMutation()  # override

需要遵守共享结果模式的类节点突变将从 SharedResultMutation 而不是 Mutation 继承,并覆盖 mutate_and_share_result 而不是 mutate

class UpsertParent(SharedResultMutation, ParentType):
    class Arguments:
        data = ParentInput()

    @staticmethod
    def mutate_and_share_result(root: None, info: graphene.ResolveInfo, data: ParentInput, *___, **____):
        return UpsertParent(id=1, name="test")  # <-- example

类边突变需要访问shared_results字典,所以他们直接覆盖mutate

class AddSibling(SharedResultMutation):
    class Arguments:
        node1 = graphene.String(required=True)
        node2 = graphene.String(required=True)

    ok = graphene.Boolean()

    @staticmethod
    def mutate(root: None, info: graphene.ResolveInfo, shared_results: dict, node1: str, node2: str):  # ISSUE: this breaks type awareness
        node1_ : ChildType = shared_results.get(node1)
        node2_ : ChildType = shared_results.get(node2)
        # do stuff
        return AddSibling(ok=True)

基本上就是这样(其余的是常见的石墨烯样板和测试模型)。我们现在可以执行如下查询:

mutation ($parent: ParentInput, $child1: ChildInput, $child2: ChildInput) {
    n1: upsertParent(data: $parent) {
        pk
        name
    }

    n2: upsertChild(data: $child1) {
        pk
        name
    }

    n3: upsertChild(data: $child2) {
        pk
        name
    }

    e1: setParent(parent: "n1", child: "n2") { ok }

    e2: setParent(parent: "n1", child: "n3") { ok }

    e3: addSibling(node1: "n2", node2: "n3") { ok }
}

问题是类似边的突变参数不满足 GraphQL 提倡的 类型意识:在 GraphQL 精神中,node1node2 应键入 graphene.Field(ChildType),而不是此实现中的 graphene.String()编辑 Added basic type checking for edge-like mutation input nodes.


编辑 2:嵌套创作

为了比较,我还实现了一个只解析创建的嵌套模式(这是我们在之前的查询中无法获得数据的唯一情况),one-file program available on Github.

它是 classic 石墨烯,除了突变 UpsertChild 我们添加字段来解决嵌套创建 他们的解析器:

class UpsertChild(graphene.Mutation, ChildType):
    class Arguments:
        data = ChildInput()

    create_parent = graphene.Field(ParentType, data=graphene.Argument(ParentInput))
    create_sibling = graphene.Field(ParentType, data=graphene.Argument(lambda: ChildInput))

    @staticmethod
    def mutate(_: None, __: graphene.ResolveInfo, data: ChildInput):
        return Child(
            pk=data.pk
            ,name=data.name
            ,parent=FakeParentDB.get(data.parent)
            ,siblings=[FakeChildDB[pk] for pk in data.siblings or []]
        )  # <-- example

    @staticmethod
    def resolve_create_parent(child: Child, __: graphene.ResolveInfo, data: ParentInput):
        parent = UpsertParent.mutate(None, __, data)
        child.parent = parent.pk
        return parent

    @staticmethod
    def resolve_create_sibling(node1: Child, __: graphene.ResolveInfo, data: 'ChildInput'):
        node2 = UpsertChild.mutate(None, __, data)
        node1.siblings.append(node2.pk)
        node2.siblings.append(node1.pk)
        return node2

所以额外的stuff的数量与节点+边模式相比是很小的。我们现在可以执行如下查询:

mutation ($parent: ParentInput, $child1: ChildInput, $child2: ChildInput) {
    n1: upsertChild(data: $child1) {
        pk
        name
        siblings { pk name }

        parent: createParent(data: $parent) { pk name }

        newSibling: createSibling(data: $child2) { pk name }
    }
}

然而,我们可以看到,与节点+边模式的可能性相比,(shared_result_mutation.py)我们不能在同一个突变中设置新兄弟的父节点。显而易见的原因是我们没有它的数据(特别是它的 pk)。另一个原因是因为不能保证嵌套突变的顺序。因此不能创建,例如,一个无数据的突变 assignParentToSiblings,它将设置当前 root 子节点的所有兄弟节点的父节点,因为嵌套的兄弟节点可能在嵌套父级。

虽然在某些实际情况下,我们只需要创建一个新对象并 然后 link 它到一个现有的对象。嵌套可以满足这些用例。


问题的评论中有人建议使用 嵌套数据 进行突变。这实际上是我第一次实现该功能,出于安全考虑我放弃了它。权限检查使用装饰器并且看起来像(我真的没有 Book mutations):

class UpsertBook(common.mutations.MutationMixin, graphene.Mutation, types.Book):
    class Arguments:
        data = types.BookInput()

    @staticmethod
    @authorize.grant(authorize.admin, authorize.owner, model=models.Book)
    def mutate(_, info: ResolveInfo, data: types.BookInput) -> 'UpsertBook':
        return UpsertBook(**data)  # <-- example

我认为我不应该在另一个地方进行此检查,例如在另一个具有嵌套数据的突变中。此外,在另一个突变中调用此方法需要在突变模块之间导入,我认为这不是一个好主意。我真的认为解决方案应该依赖于 GraphQL 解析功能,这就是我研究嵌套突变的原因,这让我首先提出了这个 post 的问题。

此外,我对问题中的 uuid 想法进行了更多测试(使用单元测试 Tescase)。事实证明,快速连续调用 python uuid.uuid4 可能会发生冲突,因此我放弃了这个选项。

因此,我创建了 graphene-chain-mutation Python package to work with Graphene-python 并允许在同一查询中引用类边突变中的类节点突变结果。我将在下面粘贴用法部分:

5 个步骤(有关可执行示例,请参阅 test/fake.py module)。

  1. 安装包(需要graphene
pip install graphene-chain-mutation
  1. 通过继承 ShareResult before graphene.Muation:
  2. 编写 node-like 突变
 import graphene
 from graphene_chain_mutation import ShareResult
 from .types import ParentType, ParentInput, ChildType, ChildInput

 class CreateParent(ShareResult, graphene.Mutation, ParentType):
     class Arguments:
         data = ParentInput()

     @staticmethod
     def mutate(_: None, __: graphene.ResolveInfo,
                data: ParentInput = None) -> 'CreateParent':
         return CreateParent(**data.__dict__)

 class CreateChild(ShareResult, graphene.Mutation, ChildType):
     class Arguments:
         data = ChildInput()

     @staticmethod
     def mutate(_: None, __: graphene.ResolveInfo,
                data: ChildInput = None) -> 'CreateChild':
         return CreateChild(**data.__dict__)
  1. 通过继承 ParentChildEdgeMutation(对于 FK 关系)或 SiblingEdgeMutation(对于 m2m 关系)创建 边缘状 突变。指定其输入节点的类型并实现set_link方法:
 import graphene
 from graphene_chain_mutation import ParentChildEdgeMutation, SiblingEdgeMutation
 from .types import ParentType, ChildType
 from .fake_models import FakeChildDB

 class SetParent(ParentChildEdgeMutation):

     parent_type = ParentType
     child_type = ChildType

     @classmethod
     def set_link(cls, parent: ParentType, child: ChildType):
         FakeChildDB[child.pk].parent = parent.pk

 class AddSibling(SiblingEdgeMutation):

     node1_type = ChildType
     node2_type = ChildType

     @classmethod
     def set_link(cls, node1: ChildType, node2: ChildType):
         FakeChildDB[node1.pk].siblings.append(node2.pk)
         FakeChildDB[node2.pk].siblings.append(node1.pk)
  1. 照常创建架构
 class Query(graphene.ObjectType):
     parent = graphene.Field(ParentType, pk=graphene.Int())
     parents = graphene.List(ParentType)
     child = graphene.Field(ChildType, pk=graphene.Int())
     children = graphene.List(ChildType)

 class Mutation(graphene.ObjectType):
     create_parent = CreateParent.Field()
     create_child = CreateChild.Field()
     set_parent = SetParent.Field()
     add_sibling = AddSibling.Field()

 schema = graphene.Schema(query=Query, mutation=Mutation)
  1. 执行查询时指定ShareResultMiddleware中间件:
 result = schema.execute(
     GRAPHQL_MUTATION
     ,variables = VARIABLES
     ,middleware=[ShareResultMiddleware()]
 )

现在GRAPHQL_MUTATION可以是一个查询,其中类边突变引用类节点突变的结果:

GRAPHQL_MUTATION = """
mutation ($parent: ParentInput, $child1: ChildInput, $child2: ChildInput) {
    n1: upsertParent(data: $parent) {
        pk
        name
    }

    n2: upsertChild(data: $child1) {
        pk
        name
    }

    n3: upsertChild(data: $child2) {
        pk
        name
    }

    e1: setParent(parent: "n1", child: "n2") { ok }

    e2: setParent(parent: "n1", child: "n3") { ok }

    e3: addSibling(node1: "n2", node2: "n3") { ok }
}
"""

VARIABLES = dict(
    parent = dict(
        name = "Emilie"
    )
    ,child1 = dict(
        name = "John"
    )
    ,child2 = dict(
        name = "Julie"
    )
)