如何使用带有apollo-upload-client的graphene-file-upload将graphql中的文件上传到Python数据库并在前端做出反应。？

Posted 2023-03-08

【中文标题】如何使用带有apollo-upload-client的graphene-file-upload将graphql中的文件上传到Python数据库并在前端做出反应。？

【英文标题】：How to Upload files in graphql using graphene-file-upload with apollo-upload-client to Python Database and react front-end.?

【发布时间】：2020-01-08 18:44:00

【问题描述】：

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

这是我的一些代码。

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

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

我的 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)

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

【参考方案1】：

现在我已经在 GraphQLView 中覆盖了 parse_body，以便它正确处理 multipart/form-data。

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

【参考方案2】：

要求

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

客户端 -

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)

运行-

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

服务器端 -

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)