mutation.py

import six
import itertools

from attrdict import AttrDict
import graphene
from graphene.core.classtypes.objecttype import ObjectTypeOptions
from graphene.core.types import ArgumentsGroup
from graphene.core.types.definitions import NonNull
from graphene.core.classtypes.mutation import Mutation, MutationMeta

from .serializer_converter import convert_serializer_class
from .types import ErrorType


class SerializerMutationOptions(ObjectTypeOptions):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, serializer_class=None, **kwargs)


class SerializerMutationMeta(MutationMeta):
    options_class = SerializerMutationOptions

    def construct(cls, bases, attrs):
        serializer_class = cls._meta.serializer_class
        if serializer_class:
            cls.add_to_class('arguments',
                             cls.construct_arguments(serializer_class))

        return super().construct(bases, attrs)

    def construct_arguments(cls, serializer_class):
        new_input_type = convert_serializer_class(serializer_class)
        cls.add_to_class('input_type', new_input_type)
        return ArgumentsGroup(input=NonNull(new_input_type))


class SerializerMutation(six.with_metaclass(SerializerMutationMeta,
                                            Mutation)):
    errors = graphene.List(ErrorType,
                           description='May contain more than one error for '
                                       'same field.')

    class Meta:
        abstract = True

    @classmethod
    def mutate(cls, instance, args, info):
        # noinspection PyShadowingBuiltins
        input = args.get('input')
        serializer = cls._meta.serializer_class(data=dict(input))
        if serializer.is_valid():
            return cls.perform_mutate(serializer, info)
        else:
            return cls(errors=itertools.chain(
                *[[AttrDict(field=k, message=v) for v in vl]
                  for k, vl in serializer.errors.items()]
            ))

    @classmethod
    def perform_mutate(cls, serializer, info):
        raise NotImplementedError()

serializer_converter.py

from functools import singledispatch

from django.core.exceptions import ImproperlyConfigured
from graphene import Boolean, Float, Int, String, NonNull
from graphene.core.classtypes import InputObjectType
from rest_framework import serializers


def convert_serializer_class(serializer_class):
    serializer = serializer_class()
    items = {name: convert_serializer_field(field)
             for name, field in serializer.fields.items()}
    return type('{}Input'.format(serializer.__class__.__name__),
                (InputObjectType,),
                items)


@singledispatch
def convert_serializer_field(field):
    raise ImproperlyConfigured(
        "Don't know how to convert the serializer field %s (%s) "
        "to Graphene type" %
        (field, field.__class__)
    )


def required_to_non_null(func):
    def wrap(field):
        graphql_type = func(field)
        return NonNull(graphql_type) if field.required else graphql_type
    return wrap


@convert_serializer_field.register(serializers.CharField)
@convert_serializer_field.register(serializers.EmailField)
@convert_serializer_field.register(serializers.URLField)
@convert_serializer_field.register(serializers.RegexField)
@convert_serializer_field.register(serializers.SlugField)
@convert_serializer_field.register(serializers.UUIDField)
@convert_serializer_field.register(serializers.IPAddressField)
@required_to_non_null
def convert_serializer_field_to_string(field):
    return String(description=field.help_text)


@convert_serializer_field.register(serializers.IntegerField)
@required_to_non_null
def convert_serializer_field_to_int(field):
    return Int(description=field.help_text)


@convert_serializer_field.register(serializers.FloatField)
@convert_serializer_field.register(serializers.DecimalField)
@required_to_non_null
def convert_serializer_field_to_float(field):
    return Float(description=field.help_text)


@convert_serializer_field.register(serializers.BooleanField)
@convert_serializer_field.register(serializers.NullBooleanField)
@required_to_non_null
def convert_serializer_field_to_boolean(field):
    return Boolean(description=field.help_text)