Python graphene 模块，Mutation() 实例源码

def test_gen_mutation_user_mutate_wrong_return(schema_builder, mock_person):
    import graphene

    from graphene_mongo.mutation import gen_mutation
    from graphene_mongo.model import ModelSchema

    def mutate(args, context):
        return False

    model_schema = ModelSchema(mock_person, mock_person._fields, mutate, None)

    result = gen_mutation(mock_person, model_schema.schema, model_schema.operators_mutation,
                          model_schema.fields_mutation, model_schema.mutate_func, None)

    assert issubclass(result, graphene.Mutation)
    assert hasattr(result, 'mutate')

    with pytest.raises(Exception) as e_info:
        result.mutate(None, {'name': "Test"}, None, None)

    assert str(e_info.value) == 'Failed to resolve mutation of the schema {}' \
                                ' because mutate function must return a instance of {}, and the return type was {}.'\
                                .format(model_schema.schema.__name__, model_schema.model.__name__, type(False))

def test_gen_mutation(mock_person):
    import inspect
    import graphene
    from graphene.utils.str_converters import to_snake_case

    from graphene_mongo.mutation import gen_mutation
    from graphene_mongo.model import ModelSchema

    model_schema = ModelSchema(mock_person, mock_person._fields, None, None)

    result = gen_mutation(mock_person, model_schema.schema, model_schema.operators_mutation,
                          model_schema.fields_mutation, None, None)

    assert issubclass(result, graphene.Mutation)
    assert hasattr(result, 'mutate')

    assert result._meta.name == 'Create' + mock_person.__name__
    assert result._meta.local_fields[to_snake_case(mock_person.__name__)]
    assert result._meta.fields[to_snake_case(mock_person.__name__)]

    operators_mutation = inspect.signature(result.__dict__['Field']).parameters['args'].default

    assert operators_mutation == model_schema.operators_mutation

def test_gen_mutation_user_mutation_func(mock_person):
    import graphene
    from graphene_mongo.mutation import gen_mutation
    from graphene_mongo.model import ModelSchema

    def mutate(args, context):
        u = mock_person(**args)
        u.save()
        return u

    model_schema = ModelSchema(mock_person, mock_person._fields, mutate, None)

    user_mutate_func = gen_mutation(mock_person, model_schema.schema, model_schema.operators_mutation,
                          model_schema.fields_mutation, mutate, None)

    assert issubclass(user_mutate_func, graphene.Mutation)
    assert hasattr(user_mutate_func, 'mutate')
    assert getattr(user_mutate_func, 'mutate').__name__ == 'user_mutate'

def mutate(cls, root, context, **kwargs):
        """
        Mutation logic is handled here
        """
        # 'before_insert' hooks won't work with db.session.bulk_save_objects
        # we need to find a way to get hooks to work with bulk_save_objects @todo

        objs = []

        for data in kwargs.get('records', []):
            d = Deal.get_object_from_graphql_input(data)
            db.session.add(d)
            objs.append(d)
        try:
            db.session.commit()
            return cls(ok=True, ids=[obj.id for obj in objs])
        except Exception as e:
            raise GraphQLError(e.args)

def mutate(cls, root, context, **kwargs):
        """ 
        Mutation logic is handled here
        """

        # More details about synchronize_session options in SqlAlchemy
        # http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.delete

        query = Deal.query.filter(
            Deal.id.in_(kwargs.get('uids', [])))

        objs = []

        for obj in query:
            db.session.delete(obj)
            objs.append(obj)

        db.session.info['changes'] = {'created': [], 'updated': [], 'deleted': objs}
        try:
            db.session.commit()
            return cls(ok=True)

        except Exception as e:
            raise GraphQLError(e.args)

def mutate(cls, root, context, **kwargs):
        """
        Mutation logic is handled here
        """
        # 'before_insert' hooks won't work with db.session.bulk_save_objects
        # we need to find a way to get hooks to work with bulk_save_objects @todo

        objs = []

        for data in kwargs.get('records', []):
            c = Contact.get_object_from_graphql_input(data)
            db.session.add(c)
            objs.append(c)
        try:
            db.session.commit()
            return cls(ok=True, ids=[obj.id for obj in objs])
        except Exception as e:
            raise GraphQLError(e.args)

def mutate(cls, root, context, **kwargs):
        """
        Mutation logic is handled here
        """

        # 'before_insert' hooks won't work with db.session.bulk_save_objects
        # we need to find a way to get hooks to work with bulk_save_objects @todo

        records = []
        for data in kwargs.get('records', []):
            actual = Deal.query.get(data['uid'])

            if not actual:
                raise GraphQLError('Invalid id (%s)' % data['id'])

            d = Deal.get_object_from_graphql_input(data)

            for column_name, _ in inspect(Deal).attrs.items():
                if column_name == 'id':
                    continue
                if column_name not in data:
                    continue
                setattr(actual, column_name, getattr(d, column_name))
            db.session.add(actual)
            records.append(actual.id)
        try:
            db.session.commit()
            return cls(ok=True, ids=records)
        except Exception as e:
            raise GraphQLError(e.args)

def mutate(cls, root, context, **kwargs):
        """
        Mutation logic is handled here
        """

        # 'before_insert' hooks won't work with db.session.bulk_save_objects
        # we need to find a way to get hooks to work with bulk_save_objects @todo

        records = []
        for data in kwargs.get('records', []):
            actual = Contact.query.get(data['uid'])

            if not actual:
                raise GraphQLError('Invalid id (%s)' % data['id'])

            c = Contact.get_object_from_graphql_input(data)

            for column_name, _ in inspect(Contact).attrs.items():
                if column_name == 'id':
                    continue
                if column_name not in data:
                    continue
                setattr(actual, column_name, getattr(c, column_name))
            db.session.add(actual)
            records.append(actual.id)
        try:
            db.session.commit()
            return cls(ok=True, ids=records)
        except Exception as e:
            raise GraphQLError(e.args)

def gen_mutation(model, graphene_schema, operators_mutation, fields_mutation, mutate_func, validator):
    """ We need to create a class that seems as follows (http://docs.graphene-python.org/en/latest/types/mutations/):

    class CreatePerson(graphene.Mutation):
        class Input:
            name = graphene.String()

        ok = graphene.Boolean()
        person = graphene.Field(lambda: Person)

        @staticmethod
        def mutate(root, args, context, info):
            person = Person(name=args.get('name'))
            ok = True
            return CreatePerson(person=person, ok=ok) 
    """

    def user_mutate(root, args, context, info):
        if validator:
            validator(model, args, {}, {})

        obj = mutate_func(args, context)
        if not isinstance(obj, model):
            raise TypeError('Failed to resolve mutation of the schema {}'
                            ' because mutate function must return a instance of {}, and the return type was {}.'
                            .format(graphene_schema.__name__, model.__name__, type(obj)))

        graphene_obj = mongo_to_graphene(obj, graphene_schema, fields_mutation)
        return Create(**{to_snake_case(model.__name__): graphene_obj})

    def generic_mutate(root, args, context, info):
        if validator:
            validator(model, args, {}, {})

        obj = model(**args)
        obj.save()
        graphene_obj = mongo_to_graphene(obj, graphene_schema, fields_mutation)
        return Create(**{to_snake_case(model.__name__): graphene_obj})

    Create = type('Create' + model.__name__, (graphene.Mutation,), {
        'Input': type('Input', (), operators_mutation),
        to_snake_case(model.__name__): graphene.Field(lambda: graphene_schema),
        'mutate': staticmethod(generic_mutate) if not mutate_func else staticmethod(user_mutate)
    })

    return Create