python - 如何使用 graphene-file-upload 和 apollo-upload-client 将 graphql 中的文件上传到 Python 数据库并在前端使用react。

Question

我正在尝试使用 graphene-file-upload 将文件上传到 django 后端，该文件具有将后端链接到 React 前端的突变，我正在尝试使用 apollo-upload 客户端将其与 graphql 链接。 在我的 django 模型中，一个空文件已成功上传，但它没有上传我选择的真实文件，而是上传一个空文件。 就像它不上传任何内容{}，但实例是在数据库中创建的，其中添加了另一个空故事。

这是我的一些代码。

我的数据库模型。模型.py

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

我的架构.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)

我的前端 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

它现在对我有用，我已经重写了 GraphQLView 中的 parse_body，以便它能够正确处理多部分/表单数据。

# 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:]]