Fastapi: File upload with fastapi graphql

[leminhson2398]

Description
Hi!
How can i upload file with the fastapi graphql
Django and flask supports a 3rd parties lib but it does not work with fastapi graphql

Additional context

[prostomarkeloff]

FastAPI (and starlette) just use graphene that have some libraries for working with files (like https://github.com/lmcgartland/graphene-file-upload, but idk how it works, the 1st answer in google). It means that you can use them. Also i think that starlette has got the 3rd party libraries for it.

[leminhson2398]

unfortunately, i tried them many ways, by googling it, searching it on github and copied other guy's code. Any way, https://github.com/lmcgartland/graphene-file-upload seem does not work with fastapi. Or maybe i did not configure it properly

[leminhson2398]

@prostomarkeloff, thank you for helping me
But i changed some code in fastapi and now it works:

elif "query" in request.query_params:
          data = request.query_params
elif "multipart/form-data" in content_type:
          form = await request.form()
          form: dict = dict(form)
          data: dict = json.loads(form.pop('operations', None))
          map_: dict = json.loads(form.pop('map', None))
          variables: dict = data.get('variables', None)
          if variables is not None:
                for key in variables.keys():
                # check if field is None or list of Nones
                if not bool(variables[key]) or not all(variables[key]):
                      varValue: list = [
                             form.get(k, None) for k in map_.keys()
                      ]
                      variables.update({
                             key: varValue
                      })
                      data.update({
                             'variables': variables
                      })

[tiangolo]

Thanks for the help here @prostomarkeloff ! :clap: :bow:

Thanks for reporting back and closing the issue @leminhson2398 :+1:

[thobiasn]

@leminhson2398 I am also looking for a way to support file upload. Where did you perform these changes? Are you still using graphene-file-upload?

[leminhson2398]

Hi @thobiasn !
I just customize the source for the class

from starlette.graphql import GraphQLApp

to this:

class CustomGraphqlApp(GraphQLApp):

    async def handle_graphql(self, request: Request) -> Response:
        if request.method in ("GET", "HEAD"):
            if "text/html" in request.headers.get("Accept", ""):
                if not self.graphiql:
                    return PlainTextResponse(
                        "Not Found", status_code=status.HTTP_404_NOT_FOUND
                    )
                return await self.handle_graphiql(request)

            # type: typing.Mapping[str, typing.Any]
            data = request.query_params

        elif request.method == "POST":
            content_type = request.headers.get("Content-Type", "")

            if "application/json" in content_type:
                data = await request.json()
                # data gonna look like this:
                """
        {
            'operationName': 'Signin',
            'variables': {'email': [email protected]', 'password': 'something'},
            'query': 'mutation Signin($email: String!, $password: String!) {\n  signin(email: $email, password: $password) {\n    ok\n    errorList\n    token\n   __typename\n  }\n}\n'
        }
        """
            elif "application/graphql" in content_type:
                body = await request.body()
                text = body.decode()
                data = {"query": text}
            elif "query" in request.query_params:
                data = request.query_params
            elif "multipart/form-data" in content_type:
                form = await request.form()
                # form looks like this:
                """{
                    'operations': '{
                        "operationName":"UploadFile",
                        "variables":{"file":null},
                        "query":"mutation UploadFile($file: Upload!) {\\n  uploadFile(file: $file) {\\n    ok\\n   errors\\n    __typename\\n  }\\n}\\n"
                    }',
                    'map': '{"1":["variables.file"]}',
                    '1': <starlette.datastructures.UploadFile object at 0x0000025395E7C1C8>,
                    ...
                }
            """
                form: dict = dict(form)
                data: dict = json.loads(form.pop('operations', None))
                map_: dict = json.loads(form.pop('map', None))
                variables: dict = data.get('variables', None)
                if variables is not None:
                    for key in variables.keys():
                        # check if field is None or list of Nones
                        if not bool(variables[key]) or not all(variables[key]):
                            varValue: list = [
                                form.get(k, None) for k in map_.keys()
                            ]
                            variables[key] = varValue
                            data.update({
                                'variables': variables
                            })
            else:
                return PlainTextResponse(
                    "Unsupported Media Type",
                    status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                )

        else:
            return PlainTextResponse(
                "Method Not Allowed", status_code=status.HTTP_405_METHOD_NOT_ALLOWED
            )

        try:
            query = data["query"]
            variables = data.get("variables")
            operation_name = data.get("operationName")
        except KeyError:
            return PlainTextResponse(
                "No GraphQL query found in the request",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        background = BackgroundTasks()
        context = {"request": request, "background": background}

        result = await self.execute(
            query, variables=variables, context=context, operation_name=operation_name
        )
        error_data = (
            [format_graphql_error(err) for err in result.errors]
            if result.errors
            else None
        )
        response_data = {"data": result.data, "errors": error_data}
        status_code = (
            status.HTTP_400_BAD_REQUEST if result.errors else status.HTTP_200_OK
        )

        return JSONResponse(
            response_data, status_code=status_code, background=background
        )

Then change the code for the api route:

from fastapi import FastAPI
from graphql.execution.executors.asyncio import AsyncioExecutor


app = FastAPI()

app.add_route(
    path='/',
    route=CustomGraphqlApp(
        schema=Schema(query=Query, mutation=Mutation),
        executor_class=AsyncioExecutor
    )
)

It worked for me that time.
You should consider this code before using it since i wrote it long ago and was not sure it is good.
Best Regards!

[thobiasn]

Hey leminhson.

Thanks so much for answering my question! Just reopening this real quick as it looks like you might have accidentally forgot to remove some comments in your code snippet and it looks like there is user information in the comments. Just making sure you are aware of this/that the password shown is not a valid one.

[leminhson2398]

Hey @thobiasn brother!
I really love helping other people if i can. Some programmers from your country or from the US also supported me a lot.
Yes you are right, i forgot to remove these private information. many thanks for reminding me.

By the way, Did my solution help you solve your issue ? I would love to hear from you.

best wishes!

[thobiasn]

Hey again @leminhson2398.
Glad I was able to help you as well. Your solution was exactly the missing piece I needed and I can now upload multipart through graphql, many many thanks!

All the best for you in the future, my friend.

[leminhson2398]

@thobiasn hey!
Glad to here that my code was useful. However i suggest you to keep an eye on it, since that code was not tested for production use. I wonder if i can customise it easily but why didn't fastapi's developers implement it ? Graphene python also doesn't implement. And we have to use a 3rd party library.

[amnan181]

Thank you so much @leminhson2398

[leminhson2398]

@amnan181
No problem