pythonflaskgraphqlariadne-graphql

Upload file with GraphQL +Ariadne + Flask: graphql.error.graphql_error.GraphQLError: Operation data should be a JSON object


I'm trying to upload a file using this technology stack (Flask + Graphql + Ariadne). I've implemented other APIs but now for file upload it is returning this error.

ERROR

Traceback (most recent call last):
  File "d:\project\python-envs\graph_ql_env\lib\site-packages\ariadne\graphql.py", line 138, in graphql_sync
    validate_data(data)
  File "d:\project\python-envs\graph_ql_env\lib\site-packages\ariadne\graphql.py", line 354, in validate_data
    raise GraphQLError("Operation data should be a JSON object")
graphql.error.graphql_error.GraphQLError: Operation data should be a JSON object
127.0.0.1 - - [Time] "POST /graphql HTTP/1.1" 400 -

The project files.

schema.graphql

scalar Upload

type Mutation {
    uploadProfileImage(id: String!, image: Upload!): Boolean!
}

mutation.py

@convert_kwargs_to_snake_case
def resolve_upload_profile_image(obj, info, id, file):

main.py

mutation.set_field("uploadProfileImage", M.resolve_upload_profile_image)

type_defs = load_schema_from_path("schema.graphql")
schema = make_executable_schema(
    type_defs, query, mutation, upload_scalar, snake_case_fallback_resolvers
)

@app.route("/graphql", methods=["POST"])
def graphql_server():
    data = request.get_json()

    success, result = graphql_sync(
        schema,
        data,
        context_value=request,
        debug=app.debug
    )

    status_code = 200 if success else 400
    return jsonify(result), status_code

Querying with Altair

mutation ($image: Upload!) {
  uploadProfileImage (id:"qwe1", image:$image)
}

Solution

  • Found a solution here: https://spectrum.chat/ariadne/general/does-anyone-have-a-working-example-of-file-uploads-in-ariadne~e5fee18b-6e41-4669-b723-6526cd1c549d

    @graphql_bp.route("/", methods=["POST"])
    def graphql_server():
        if request.content_type.startswith("multipart/form-data"):
            data = combine_multipart_data(
                json.loads(request.form.get("operations")),
                json.loads(request.form.get("map")),
                dict(request.files)
            )
        else:
            data = request.get_json()
        success, result = graphql_sync(
            schema,
            data,
            context_value=request,
            debug=current_app.debug
        )
        status_code = 200 if success else 400
        return jsonify(result), status_code