Geekly articles weekly

石墨烯-Python历险年

石墨烯-Python历险年


图片


大家好，我是python开发人员。 去年，我使用了graphene-python + django ORM，在这段时间里，我尝试创建某种工具来简化使用石墨烯的工作。 结果，我得到了一个小的graphene-framework代码库和一组我想分享的规则。


图片


什么是石墨烯-python？


根据graphene-python.org ，则：


Graphene-Python是用于使用Python轻松创建GraphQL API的库。 它的主要任务是提供一个简单但同时可扩展的API，以简化程序员的生活。

它的主要任务是提供一个简单但同时可扩展的API，以简化程序员的生活。


是的，实际上石墨烯是简单且可扩展的，但在我看来，对于大型且快速增长的应用来说，它太简单了。 简短的文档（我改用了源代码-更为冗长），加上缺少编写代码的标准，因此该库不是您下一个API的最佳选择。


无论如何，幸运的是，我决定在项目中使用它，并解决了大多数问题（由于石墨烯具有丰富的未记录特征）。 我的某些解决方案纯粹是架构性的，无需框架即可直接使用。 但是，它们的其余部分仍需要一些代码库。


本文不是文档，而是从某种意义上简短地描述了我所走的道路以及以一种或多种方式解决的问题，并简要说明了选择的理由。 在这一部分中，我关注突变及其相关的内容。


本文的目的是获得任何有意义的反馈 ，因此我将等待评论中的批评！


注意：在继续阅读本文之前，强烈建议您熟悉GraphQL。



变异


关于GraphQL的大多数讨论都集中在获取数据上，但是任何自重平台都需要一种方法来修改存储在服务器上的数据。

让我们从突变开始。


考虑以下代码：


 class UpdatePostMutation(graphene.Mutation): class Arguments: post_id = graphene.ID(required=True) title = graphene.String(required=True) content = graphene.String(required=True) image_urls = graphene.List(graphene.String, required=False) allow_comments = graphene.Boolean(required=True) contact_email = graphene.String(required=True) ok = graphene.Boolean(required=True) errors = graphene.List(graphene.String, required=True) def mutate(_, info, post_id, title, content, image_urls, allow_comments, contact_email): errors = [] try: post = get_post_by_id(post_id) except PostNotFound: return UpdatePostMutation(ok=False, errors=['post_not_found']) if not info.context.user.is_authenticated: errors.append('not_authenticated') if len(title) < TITLE_MIN_LENGTH: errors.append('title_too_short') if not is_email(contact_email): errors.append('contact_email_not_valid') if post.owner != info.context.user: errors.append('not_post_owner') if Post.objects.filter(title=title).exists(): errors.append('title_already_taken') if not errors: post = Utils.update_post(post, title, content, image_urls, allow_comments, contact_email) return UpdatePostMutation(ok=bool(errors), errors=errors)

UpdatePostMutation使用传输的数据更改具有给定id的帖子，并在不满足某些条件的情况下返回错误。


只需查看一下此代码，由于以下原因，它的不可扩展性和不可支持性就变得可见：


  1. mutate函数的参数太多，即使我们想添加更多要编辑的字段，其数量也会增加。
  2. 为了使突变在客户端看起来相同，它们必须返回errors和ok ，以便它们的状态以及始终可以理解的内容。
  3. 在mutate函数中搜索和检索对象 。 禁食中有突变功能，如果不存在，则不应发生突变。
  4. 检查变异中的权限。 如果用户无权执行此操作（编辑某些帖子），则不应发生突变。
  5. 一个无用的第一个参数（对于顶级字段，始终为None的根 ，这是我们的变体）。
  6. 一组无法预测的错误：如果您没有源代码或文档，那么您将不知道此突变会返回哪些错误，因为它们不会反映在方案中。
  7. 直接在mutate方法中执行的模板错误检查太多，这涉及更改数据，而不是各种检查。 理想的mutate应该由一行组成-调用后编辑功能。

简而言之， mutate应该修改data ，而不要处理第三方任务，例如访问对象和验证输入。 我们的目标是达到以下目标：


  def mutate(post, info, input): post = Utils.update_post(post, **input) return UpdatePostMutation(post=post)

现在让我们看一下以上几点。



自定义类型


email字段是作为字符串传递的，而它是特定格式的字符串 。 API每次收到电子邮件时，都必须检查其正确性。 因此，最好的解决方案是创建一个自定义类型。


 class Email(graphene.String): # ...

这似乎很明显，但值得一提。



输入类型


使用输入类型进行突变。 即使它们不会在其他地方重复使用。 由于输入类型的原因，查询变得更小，从而使它们更易于阅读和编写。


 class UpdatePostInput(graphene.InputObjectType): title = graphene.String(required=True) content = graphene.String(required=True) image_urls = graphene.List(graphene.String, required=False) allow_comments = graphene.Boolean(required=True) contact_email = graphene.String(required=True)

至：


 mutation( $post_id: ID!, $title: String!, $content: String!, $image_urls: String!, $allow_comments: Boolean!, $contact_email: Email! ) { updatePost( post_id: $post_id, title: $title, content: $content, image_urls: $image_urls, allow_comments: $allow_comments, contact_email: $contact_email, ) { ok } }

之后：


 mutation($id: ID!, $input: UpdatePostInput!) { updatePost(id: $id, input: $input) { ok } }

突变代码更改为：


 class UpdatePostMutation(graphene.Mutation): class Arguments: input = UpdatePostInput(required=True) id = graphene.ID(required=True) ok = graphene.Boolean(required=True) errors = graphene.List(graphene.String, required=True) def mutate(_, info, input, id): # ... if not errors: post = Utils.update_post(post, **input.__dict__) return UpdatePostMutation(errors=errors)


基础突变类别


如第2段所述， 突变必须返回errors并可以ok以便始终可以了解其状态和原因 。 这很简单，我们创建一个基类：


 class MutationPayload(graphene.ObjectType): ok = graphene.Boolean(required=True) errors = graphene.List(graphene.String, required=True) query = graphene.Field('main.schema.Query', required=True) def resolve_ok(self, info): return len(self.errors or []) == 0 def resolve_errors(self, info): return self.errors or [] def resolve_query(self, info): return {}

一些注意事项：


  • resolve_ok方法已resolve_ok ，因此我们不必自己计算ok 。
  • query字段是根Query ，它使您可以直接在变异请求中查询数据（变异完成后将请求数据）。
     mutation($id: ID!, $input: PostUpdateInput!) { updatePost(id: $id, input: $input) { ok query { profile { totalPosts } } } }

当客户端在突变完成后更新一些数据并且不想让支持者返回整个集合时，这非常方便。 您编写的代码越少，维护起来就越容易。 我从这里接受了这个想法。


使用基本突变类，代码变为：


 class UpdatePostMutation(MutationPayload, graphene.Mutation): class Arguments: input = UpdatePostInput(required=True) id = graphene.ID(required=True) def mutate(_, info, input, id): # ...


根突变


现在，我们的变更请求如下所示：


 mutation($id: ID!, $input: PostUpdateInput!) { updatePost(id: $id, input: $input) { ok } }

在全球范围内包含所有突变不是一个好习惯。 原因如下：


  1. 随着突变数量的增加，找到所需的突变变得越来越困难。
  2. 由于有一个名称空间，因此必须在突变名称中包含“其模块的名称”，例如update Post 。
  3. 您必须将id作为参数传递给突变。

我建议使用根突变 。 他们的目标是通过将突变分成单独的范围并从访问对象和访问对象的逻辑中释放突变来解决这些问题。


新请求如下所示：


 mutation($id: ID!, $input: PostUpdateInput!) { post(id: $id) { update(input: $input) { ok } } }

请求参数保持不变。 现在，更改功能在post内部被“调用”，它允许实现以下逻辑：


  1. 如果没有将id传递到post ，那么它将返回{} 。 这使您可以继续在其中执行突变。 用于不需要根元素的突变（例如，创建对象）。
  2. 如果传递了id ，则检索相应的元素。
  3. 如果未找到对象， None返回None ，从而完成请求，不调用该突变。
  4. 如果找到该对象，则检查用户的权限以对其进行操作。
  5. 如果用户没有权限， None返回None并且请求完成，不会调用该突变。
  6. 如果用户具有权限，则返回找到的对象，并且该变异将其作为根（第一个参数）接收。

因此，突变代码更改为：


 class UpdatePostMutation(MutationPayload, graphene.Mutation): class Arguments: input = UpdatePostInput() def mutate(post, info, input): if post is None: return None errors = [] if not info.context.user.is_authenticated: errors.append('not_authenticated') if len(title) < TITLE_MIN_LENGTH: errors.append('title_too_short') if Post.objects.filter(title=title).exists(): errors.append('title_already_taken') if not errors: post = Utils.update_post(post, **input.__dict__) return UpdatePostMutation(errors=errors)

  • 现在，突变的根（第一个参数）是Post类型的对象，在该对象上执行突变。
  • 授权检查已移至根突变代码。

根突变代码：


 class PostMutationRoot(MutationRoot): class Meta: model = Post has_permission = lambda post, user: post.owner == user update = UpdatePostMutation.Field()


错误界面


为了使一组错误可预测，它们必须反映在设计中。


  • 由于变异会返回多个错误，因此错误应列为清单。
  • 由于错误由不同的类型表示，因此对于特定的突变，必须存在特定Union错误Union 。
  • 为了使错误彼此之间保持相似，它们必须实现接口，我们将其ErrorInterface 。 让它包含两个字段： ok和message 。

因此，错误必须是[SomeMutationErrorsUnion]!类型[SomeMutationErrorsUnion]! 。 SomeMutationErrorsUnion所有子类型SomeMutationErrorsUnion必须实现ErrorInterface 。


我们得到：


 class NotAuthenticated(graphene.ObjectType): message = graphene.String(required=True, default_value='not_authenticated') class Meta: interfaces = [ErrorInterface, ] class TitleTooShort(graphene.ObjectType): message = graphene.String(required=True, default_value='title_too_short') class Meta: interfaces = [ErrorInterface, ] class TitleAlreadyTaken(graphene.ObjectType): message = graphene.String(required=True, default_value='title_already_taken') class Meta: interfaces = [ErrorInterface, ] class UpdatePostMutationErrors(graphene.Union): class Meta: types = [NotAuthenticated, TitleIsTooShort, TitleAlreadyTaken, ]

看起来不错，但是代码太多。 我们使用元类动态生成这些错误：


 class PostErrors(metaclass=ErrorMetaclass): errors = [ 'not_authenticated', 'title_too_short', 'title_already_taken', ] class UpdatePostMutationErrors(graphene.Union): class Meta: types = [PostErrors.not_authenticated, PostErrors.title_too_short, PostErrors.title_already_taken, ]

将返回的错误的声明添加到突变中：


 class UpdatePostMutation(MutationPayload, graphene.Mutation): class Arguments: input = UpdatePostInput() errors = graphene.List(UpdatePostMutationErrors, required=True) def mutate(post, info, input): # ...


检查错误


在我看来，除了改变数据之外， mutate方法不应该关心任何其他事情。 为此，您需要为此功能检查其代码中的错误。


省略实现，结果如下：


 class UpdatePostMutation(DefaultMutation): class Arguments: input = UpdatePostInput() class Meta: root_required = True authentication_required = True #   ,    True   # An iterable of tuples (error_class, checker) checks = [ ( PostErrors.title_too_short, lambda post, input: len(input.title) < TITLE_MIN_LENGTH ), ( PostErrors.title_already_taken, lambda post, input: Post.objects.filter(title=input.title).exists() ), ] def mutate(post, info, input): post = Utils.update_post(post, **input.__dict__) return UpdatePostMutation()

在启动mutate函数之前， mutate调用每个检查器 （ checks数组的成员的第二个元素）。 如果返回True ，则找到相应的错误。 如果没有发现错误，则调用mutate函数。


我将解释：


  • Check函数采用与mutate函数相同的参数。
  • 如果发现错误，验证函数应返回True 。
  • 授权检查和根元素的存在非常普遍，并在Meta标志中列出。
  • authentication_required添加授权检查是否为True 。
  • root_required添加一个“ root is not None ”检查。
  • 不再需要UpdatePostMutationErrors 。 可能会根据checks数组的错误类别即时创建一个可能的错误并集。


泛型


上一节中使用DefaultMutation添加了pre_mutate方法，该方法使您可以在检查错误之前更改输入参数，并相应地调用突变。


还有一个通用入门工具包，可简化代码并使工作更轻松。
注意：当前通用代码特定于django ORM


创建变异


需要model或create_function 。 默认情况下， create_function如下所示：


 model._default_manager.create(**data, owner=user)

这可能看起来不安全，但是请不要忘记graphql中有内置的类型检查以及突变检查。


它还提供了一个post_mutate方法，该方法在create_function之后带有参数(instance_created, user)被调用，其结果将返回给客户端。


更新突变


允许您设置update_function 。 默认情况下：


 def default_update_function(instance, user=None, **data): instance.__dict__.update(data) instance.save() return instance

root_required默认为True 。


它还提供了一个post_mutate方法，该方法在update_function之后使用参数(instance_updated, user)进行调用，其结果将返回给客户端。


这就是我们所需要的！


最终代码：


 class UpdatePostMutation(UpdateMutation): class Arguments: input = UpdatePostInput() class Meta: checks = [ ( PostErrors.title_too_short, lambda post, input: len(input.title) < TITLE_MIN_LENGTH ), ( PostErrors.title_already_taken, lambda post, input: Post.objects.filter(title=input.title).exists() ), ]

DeleteMutation


允许您设置delete_function 。 默认情况下：


 def default_delete_function(instance, user=None, **data): instance.delete()


结论


本文仅考虑一个方面，尽管我认为这是最复杂的。 我对解析器和类型以及石墨烯-python中的常规内容有一些想法。


我很难称自己为经验丰富的开发人员，因此，我将很高兴收到任何反馈和建议。


源代码可以在这里找到 。

Source: https://habr.com/ru/post/zh-CN461939/

More articles:


All Articles