How to Upload files in graphql using graphene-file-upload with apollo-upload-client to Python Database and react front-end.?

Question

I'm trying to upload a file to a django backend using graphene-file-upload which has the mutation to link the backend to the react frontend where I'm trying to use apollo-upload client to link it with graphql. In my django model an empty file is being successfully uploaded but it is not uploading the real file which I'm selecting but uploads an empty file. Like it uploads nothing {} but the instance is created in the database where another story is added which is empty.

Here is some of my code.

My Database Model. models.py

class Story(models.Model):
    file = models.FileField(null=True)
    created_at = models.DateTimeField(auto_now_add=True)

MY schema.py

from graphene_file_upload.scalars import Upload

class StoryType(DjangoObjectType):
    class Meta:
        model = Story

class UploadFile(graphene.Mutation):
    story = graphene.Field(StoryType)

    class Arguments:
        file = Upload()

    def mutate(self, info, file):

        for line in file:
            print(line)

        story = Story(file=file)

        story.save()
        return UploadFile(story=story)

My frontend File.js

import React from 'react';
import { Mutation } from 'react-apollo';
import {withStyles} from '@material-ui/core/styles';
import gql from 'graphql-tag';

const styles = theme => ({
    layoutRoot: {}
});


const UploadFile = () => (
  <Mutation
    mutation={gql`
      mutation($file: Upload!) {
        uploadFile(file: $file) {
          story {
            file
          }
        }
      }
    `}
  >
    {mutate => (
      <input
        type="file"
        required
        onChange={({
          target: {
            validity,
            files: [file]
          }
        }) => validity.valid && mutate({ variables: { file } })}
      />
    )}
  </Mutation>
)

export default withStyles(styles, {withTheme: true})(UploadFile);

Answer 1

Requirements

from python_graphql_client import GraphqlClient
import asyncio
client = GraphqlClient(endpoint="http://localhost:8000/graphql")

Client-side -

def test_archive_upload(client):
    file_header = "data:application/zip;base64,"
    with open("files/Archive.zip", 'rb') as file:
    query = """
            mutation uploadFile($file: Upload) {
                uploadFile(file:$file) {
                    ok
                }
            }
        """
    file = file.read()
    file = file_header + base64.b64encode(file).decode("UTF-8")
    variables = {"file": file}
    data = client.execute(query=query, variables=variables)

Run -

asyncio.get_event_loop().run_until_complete(test_archive_upload(client))

Server-side -

file = file.split(",")
file[0] = file[0]+","
file_type = guess_extension(guess_type(file[0])[0])
file = base64.b64decode(file[1])
with open("files/test"+ file_type, "wb") as w_file:
    w_file.write(file)

Answer 2

It's working for me now I've overridden parse_body in GraphQLView, in order for it to deal with multipart/form-data correctly.

# views.py
from django.http.response import HttpResponseBadRequest
from graphene_django.views import GraphQLView

class MyGraphQLView(GraphQLView):

    def parse_body(self, request):
        content_type = self.get_content_type(request)

        if content_type == "application/graphql":
            return {"query": request.body.decode()}

        elif content_type == "application/json":
            # noinspection PyBroadException
            try:
                body = request.body.decode("utf-8")
            except Exception as e:
                raise HttpError(HttpResponseBadRequest(str(e)))

            try:
                request_json = json.loads(body)
                if self.batch:
                    assert isinstance(request_json, list), (
                        "Batch requests should receive a list, but received {}."
                    ).format(repr(request_json))
                    assert (
                        len(request_json) > 0
                    ), "Received an empty list in the batch request."
                else:
                    assert isinstance(
                        request_json, dict
                    ), "The received data is not a valid JSON query."
                return request_json
            except AssertionError as e:
                raise HttpError(HttpResponseBadRequest(str(e)))
            except (TypeError, ValueError):
                raise HttpError(HttpResponseBadRequest("POST body sent invalid JSON."))

        # Added for graphql file uploads
        elif content_type == 'multipart/form-data':
            operations = json.loads(request.POST['operations'])
            files_map = json.loads(request.POST['map'])
            return place_files_in_operations(
                operations, files_map, request.FILES)

        elif content_type in [
            "application/x-www-form-urlencoded",
            #"multipart/form-data",
        ]:
            return request.POST

        return {}

def place_files_in_operations(operations, files_map, files):
    # operations: dict or list
    # files_map: {filename: [path, path, ...]}
    # files: {filename: FileStorage}

    fmap = []
    for key, values in files_map.items():
        for val in values:
            path = val.split('.')
            fmap.append((path, key))

    return _place_files_in_operations(operations, fmap, files)


def _place_files_in_operations(ops, fmap, fobjs):
    for path, fkey in fmap:
        ops = _place_file_in_operations(ops, path, fobjs[fkey])
    return ops

def _place_file_in_operations(ops, path, obj):

    if len(path) == 0:
        return obj

    if isinstance(ops, list):
        key = int(path[0])
        sub = _place_file_in_operations(ops[key], path[1:], obj)
        return _insert_in_list(ops, key, sub)

    if isinstance(ops, dict):
        key = path[0]
        sub = _place_file_in_operations(ops[key], path[1:], obj)
        return _insert_in_dict(ops, key, sub)

    raise TypeError('Expected ops to be list or dict')

def _insert_in_dict(dct, key, val):
    return {**dct, key: val}


def _insert_in_list(lst, key, val):
    return [*lst[:key], val, *lst[key+1:]]