Flask调用FastAPI接口返回422错误问题排查求助

Flask调用FastAPI的/runquery接口时返回422错误，FastAPI日志显示：

INFO:127.0.0.1:56904 - "POST /runquery HTTP/1.1" 422 Unprocessable Entity

FastAPI接口预期接收query（值为select c2,c2,c4,c5 from table_tab）和env（值为iAT）两个参数，相关代码如下：

Flask代码 #

import requests
from flask import Flask, request, render_template

app = Flask(__name__)    

@app.route("/", methods=["GET", "POST"])
def index():
    query_id = ""
    query = ""
    status = ""
    env = ""
    if request.method == "POST":
        query = request.form["query"]
        env = request.form["env"]
        data = {"query": query, "env": env}
        print(data)
        headers = {'Content-Type': 'application/json'}
        response = requests.post("http://localhost:8000/runquery", json=data, headers=headers)
        print(response)
        if response.status_code != 200:
            return f'Error: {response.status_code}'
        else:
            print(response.json())
        response_data = response.json()
        query_id = response_data["query_id"]
        status = response_data["status"]
    return render_template("index.html", query_id=query_id, query=query, status=status, env=env)

if __name__ == "__main__":
    app.run(debug=True)

FastAPI接口代码 #

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import sqlparse
import uuid
import os
from datetime import timedelta
from rq import Queue
from redis import Redis

app = FastAPI()
r = Redis()
que = Queue(connection=r)
home_path = "./queries"  # 需根据实际环境调整路径

@app.post("/runquery")
async def runquery(query: str, env: str):
    print('.....')
    delay = 60
    parsed_query = sqlparse.parse(query)[0]
    if parsed_query.get_type() == "SELECT":
        pass
    else:
        return JSONResponse(content={"error": "Only SELECT queries are allowed"}, status_code=422)

    registry_scheduled = que.scheduled_job_registry
    registry_started = que.started_job_registry

    if len(registry_scheduled.get_job_ids()) >= 2:
        return JSONResponse(content={"Info": "GQs are already in-progress. Try after some time"})

    if len(registry_started.get_job_ids()) >= 2:
        return JSONResponse(content={"Info": "GQs are already in-progress.Try after some time"})

    # 生成唯一查询ID
    query_id = str(uuid.uuid4())
    print('query_id..........', query_id)
    directory_path = os.path.join(home_path, query_id)
    print(directory_path)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    file_path = os.path.join(directory_path, "query.sql")
    with open(file_path, "w") as f:
        f.write(query)

    job = que.enqueue_in(timedelta(seconds=delay), exec_query, query, env, query_id)
    r.set(query_id, job.id)

    # 返回查询ID给用户
    return JSONResponse(content={"query_id": query_id, "query": query, "stauts": "OK"}, 
    status_code=202)

FastAPI的runquery函数中，query和env参数默认被解析为URL查询参数，但Flask端是通过JSON请求体传递参数的，两者参数传递方式不匹配，导致FastAPI无法正确识别参数，返回422错误。

另外，FastAPI返回的JSON存在拼写错误："stauts"应为"status"，但这不是触发422的直接原因。

提供两种修复方式，任选其一即可：

方式一：修改FastAPI接口，接收JSON请求体 #

使用Pydantic模型定义请求体，让接口支持解析JSON格式的参数：

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
# 其他导入保持不变

class QueryRequest(BaseModel):
    query: str
    env: str

@app.post("/runquery")
async def runquery(request: QueryRequest):
    print('.....')
    delay = 60
    parsed_query = sqlparse.parse(request.query)[0]
    if parsed_query.get_type() == "SELECT":
        pass
    else:
        return JSONResponse(content={"error": "Only SELECT queries are allowed"}, status_code=422)

    registry_scheduled = que.scheduled_job_registry
    registry_started = que.started_job_registry

    if len(registry_scheduled.get_job_ids()) >= 2:
        return JSONResponse(content={"Info": "GQs are already in-progress. Try after some time"})

    if len(registry_started.get_job_ids()) >= 2:
        return JSONResponse(content={"Info": "GQs are already in-progress.Try after some time"})

    query_id = str(uuid.uuid4())
    print('query_id..........', query_id)
    directory_path = os.path.join(home_path, query_id)
    print(directory_path)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    file_path = os.path.join(directory_path, "query.sql")
    with open(file_path, "w") as f:
        f.write(request.query)

    job = que.enqueue_in(timedelta(seconds=delay), exec_query, request.query, request.env, query_id)
    r.set(query_id, job.id)

    # 修复拼写错误：stauts -> status
    return JSONResponse(content={"query_id": query_id, "query": request.query, "status": "OK"}, 
    status_code=202)

方式二：修改Flask请求，传递查询参数 #

若不想改动FastAPI接口，Flask端可将参数作为查询参数附在URL后：

# Flask代码中修改请求发送部分
response = requests.post(
    "http://localhost:8000/runquery",
    params={"query": query, "env": env}
)
# 移除原来的headers和json参数，FastAPI会自动解析URL中的查询参数

内容的提问来源于stack exchange，提问作者Ram