Hi everyone, I am a python developer. Last year I worked with graphene-python + django ORM and during this time I tried to create some kind of tool to make working with graphene more convenient. As a result, I got a small graphene-framework
code base and a set of some rules, which I would like to share.
What is graphene-python?
According to graphene-python.org , then:
Graphene-Python is a library for easily creating GraphQL APIs using Python. Its main task is to provide a simple but at the same time extensible API to make the life of programmers easier.
Its main task is to provide a simple but at the same time extensible API to make the life of programmers easier.
Yes, in reality graphene is simple and extensible, but it seems to me too simple for large and fast-growing applications. Short documentation (I used the source code instead - it is much more verbose), as well as the lack of standards for writing code makes this library not the best choice for your next API.
Be that as it may, I decided to use it in the project and ran into a number of problems, fortunately, having solved most of them (thanks to the rich undocumented features of graphene). Some of my solutions are purely architectural and can be used out of the box, without my framework. However, the rest of them still require some code base.
This article is not documentation, but in a sense a short description of the path that I went and the problems that I solved in one way or another with a brief justification for my choice. In this part, I paid attention to mutations and things related to them.
The purpose of this article is to get any meaningful feedback , so I will wait for criticism in the comments!
Note: before continuing reading the article, I strongly recommend that you familiarize yourself with what GraphQL is.
Most discussions about GraphQL focus on getting data, but any self-respecting platform also requires a way to modify the data stored on the server.
Let's start with mutations.
Consider the following code:
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
modifies the post with the given id
, using the transmitted data and returns errors if some conditions are not met.
One has only to look at this code, as it becomes visible its non-extensibility and unsupportability due to:
mutate
function, the number of which may increase even if we want to add more fields to be edited.errors
and ok
, so that their status and what it is always possible to understand.mutate
function. The mutation function operates with fasting, and if it is not there, then the mutation should not occur.None
for top-level fields, which is our mutation).mutate
method, which involves changing the data, rather than a variety of checks. The ideal mutate
should consist of one line - a call to the post editing function.In short, mutate
should modify data , rather than take care of third-party tasks such as accessing objects and validating input. Our goal is to arrive at something like:
def mutate(post, info, input): post = Utils.update_post(post, **input) return UpdatePostMutation(post=post)
Now let's look at the points above.
The email
field is passed as a string, while it is a string of a specific format . Each time the API receives an email, it must check its correctness. So the best solution would be to create a custom type.
class Email(graphene.String): # ...
This may seem obvious, but worth mentioning.
Use input types for your mutations. Even if they are not subject to reuse in other places. Thanks to the input types, queries become smaller, which makes them easier to read and faster to write.
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)
Before:
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 } }
After:
mutation($id: ID!, $input: UpdatePostInput!) { updatePost(id: $id, input: $input) { ok } }
The mutation code changes to:
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)
As mentioned in clause 7, mutations must return errors
and ok
so that their status and what causes it can always be understood . It's simple enough, we create a base class:
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 {}
A few notes:
resolve_ok
method is resolve_ok
, so we don’t have to calculate ok
ourselves.query
field is the root Query
, which allows you to query data directly inside the mutation request (data will be requested after the mutation is completed). mutation($id: ID!, $input: PostUpdateInput!) { updatePost(id: $id, input: $input) { ok query { profile { totalPosts } } } }
This is very convenient when the client updates some data after the mutation is complete and does not want to ask the back-end to return this entire set. The less code you write, the easier it is to maintain. I took this idea from here .
With the base mutation class, the code turns into:
class UpdatePostMutation(MutationPayload, graphene.Mutation): class Arguments: input = UpdatePostInput(required=True) id = graphene.ID(required=True) def mutate(_, info, input, id): # ...
Our mutation request now looks like this:
mutation($id: ID!, $input: PostUpdateInput!) { updatePost(id: $id, input: $input) { ok } }
Containing all mutations in a global scope is not best practice. Here are some reasons why:
update
Post
.id
as an argument to the mutation.I suggest using root mutations . Their goal is to solve these problems by separating mutations into separate scopes and freeing mutations from the logic of access to objects and access rights to them.
The new request looks like this:
mutation($id: ID!, $input: PostUpdateInput!) { post(id: $id) { update(input: $input) { ok } } }
The request arguments remain the same. Now the change function is "called" inside the post
, which allows the following logic to be implemented:
id
not passed to post
, then it returns {}
. This allows you to continue performing mutations inside. Used for mutations that do not require a root element (for example, to create objects).id
is passed, the corresponding element is retrieved.None
returned and the request completes, the mutation is not called.None
returned and the request completes, the mutation is not called.Thus, the mutation code changes to:
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
, over which the mutation is performed.Root Mutation Code:
class PostMutationRoot(MutationRoot): class Meta: model = Post has_permission = lambda post, user: post.owner == user update = UpdatePostMutation.Field()
To make a set of errors predictable, they must be reflected in the design.
Union
errors must exist for a particular mutation.ErrorInterface
. Let it contain two fields: ok
and message
.Thus, errors must be of type [SomeMutationErrorsUnion]!
. All subtypes of SomeMutationErrorsUnion
must implement ErrorInterface
.
We get:
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, ]
Looks good, but too much code. We use the metaclass to generate these errors on the fly:
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, ]
Add the declaration of the returned errors to the mutation:
class UpdatePostMutation(MutationPayload, graphene.Mutation): class Arguments: input = UpdatePostInput() errors = graphene.List(UpdatePostMutationErrors, required=True) def mutate(post, info, input): # ...
It seems to me that the mutate
method should not care about anything other than mutating the data . To achieve this, you need to check for errors in their code for this function.
Omitting the implementation, here is the result:
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()
Before the mutate
function starts, each checker is called (the second element of the members of the checks
array). If True
is returned, the corresponding error is found. If no errors are found, the mutate
function is called.
I will explain:
mutate
.True
if an error is found.Meta
flags.authentication_required
adds authorization check if True
.root_required
adds a " root is not None
" check.UpdatePostMutationErrors
no longer required. A union of possible errors is created on the fly depending on the error classes of the checks
array.DefaultMutation
used in the last section adds a pre_mutate
method, which allows you to change the input arguments before checking for errors, and, accordingly, invoking the mutation.
There is also a starter kit of generics that make the code shorter and life easier.
Note: currently the generic code is specific to django ORM
Requires one of the model
or create_function
. By default, create_function
looks like this:
model._default_manager.create(**data, owner=user)
This may look unsafe, but do not forget that there is built-in type checking in graphql, as well as checks in mutations.
Also provides a post_mutate
method, which is called after create_function
with arguments (instance_created, user)
, the result of which will be returned to the client.
Allows you to set update_function
. Default:
def default_update_function(instance, user=None, **data): instance.__dict__.update(data) instance.save() return instance
root_required
is True
by default.
It also provides a post_mutate
method that is called after update_function
with arguments (instance_updated, user)
, the result of which will be returned to the client.
And this is what we need!
Final code:
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() ), ]
Allows you to set delete_function
. Default:
def default_delete_function(instance, user=None, **data): instance.delete()
This article discusses only one aspect, although in my opinion it is the most complex. I have some thoughts on resolvers and types, as well as general things in graphene-python.
It’s hard for me to call myself an experienced developer, so I will be very glad of any feedback, as well as suggestions.
Source: https://habr.com/ru/post/461939/
All Articles