使用 GraphQL +Ariadne + Flask 上传文件：graphql.error.graphql_error.GraphQLError: Operation data should be an JSON object答案

【问题标题】：Upload file with GraphQL +Ariadne + Flask: graphql.error.graphql_error.GraphQLError: Operation data should be a JSON object使用 GraphQL +Ariadne + Flask 上传文件：graphql.error.graphql_error.GraphQLError: Operation data should be an JSON object
【发布时间】：2021-09-10 00:13:55
【问题描述】：

我正在尝试使用此技术堆栈（Flask + Graphql + Ariadne）上传文件。我已经实现了其他 API，但现在对于文件上传，它返回了这个错误。

错误

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 -

项目文件。

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

使用 Altair 查询

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

【问题讨论】：

标签： python flask graphql ariadne-graphql

【解决方案1】：

在这里找到了解决方案： 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

【讨论】：