How can I write nested GraphQL mutations in graphene?

Question

I am trying to write nested mutations in graphene, but run into trouble passing the output of one mutation to another. I would like the WriteImage resolver to able to accept its input image from the ReadImage resolver. Ultimately, WriteImage should able to accept an image from other resolvers as well, so that the user can script a pipeline.

The schema looks like this:

import SimpleITK as sitk
import graphene
from graphql.execution.executors.asyncio import AsyncioExecutor

class Image(graphene.Scalar):
    '''SimpleITK Image Type'''

    @staticmethod
    def serialize(Image):
        return Image

    @staticmethod
    def parse_literal(node):
        if isinstance(node, sitk.Image):
            return node

    @staticmethod
    def parse_value(value):
        return value

class ReadImage(graphene.Mutation):
    class Input:
        path = graphene.String()

    image = Image()

    @staticmethod
    async def mutate(cls, args, context, info):
        image = sitk.ReadImage(args.get('path'))
        return ReadImage(image=image)


class WriteImage(graphene.Mutation):
    class Input:
        image = Image()
        path = graphene.String()

    image = Image()

    @staticmethod
    async def mutate(cls, args, context, info):
        image = args.get('image')
        sitk.WriteImage(image, args.get('path'))
        return WriteImage(image=image)


class Mutation(graphene.ObjectType):
    readImage = ReadImage.Field()
    writeImage = WriteImage.Field()


schema = graphene.Schema(
    query=Query,
    mutation=Mutation,
)

executed = schema.execute(
    """
        mutation {
            writeImage(
                image: {
                    readImage(
                        path: "/path/to/image.nii"
                    ) {
                        image
                    }
                },
                path: "/Users/kasper/Desktop"
            ) {
                image
            }
        }
    """, executor=AsyncioExecutor())

print(executed.data)
print(executed.errors)

This outputs

>>> None
>>> [GraphQLError('Argument "image" has invalid value ReadImageInput.
Expected "ReadImageInput", found not an object.',)]

What does this error message mean? Why is it not an object? How should the schema be structured so that I may use output from one resolver as input to another?

Additional info: Executing the ReadImage produces the following output and allows you to see the image type:

executed = schema.execute(
    """
        mutation {
            readImage(
                path: "/path/to/image.nii"
            ) {
                image
            }
        }
    """, executor=AsyncioExecutor())

print(executed.data)
print(executed.errors)

>>> OrderedDict([('readImage', OrderedDict([('image', ::value_type *' at 0x1087b5030> >)]))])
>>> None

How can I write nested GraphQL mutations in graphene?

0 Answers