Flask异步视图返回协程对象报错求助（依赖Flask2.0.2/Werkzeug2.0.3）

错误信息 #

TypeError：视图函数未返回有效响应。返回类型必须是字符串、字典、元组、Response实例或WSGI可调用对象，但实际是一个协程。

需求与问题 #

需要在Flask中实现异步视图接口/predict，接口需返回异步函数model.transform(data_set, 'prediction')的执行结果，但尝试以下方法均报错：

• 使用async def定义异步视图并通过await等待异步操作结果；
• 使用async_helpers模块；
• 结合Flask Blueprint与asyncio的run_in_executor实现异步接口。

项目依赖：Flask-2.0.2、Werkzeug-2.0.3。

代码示例 #

@bp.route("/predict", methods=["POST"])
@pre.catch({
    "userId": Rule(callback=ParamsCallback.convert_to_int, dest="user_id"),
    "modelId": Rule(type=str, dest="model_id"),
    "database": Rule(type=str, dest="database"),
    "table": Rule(type=str, dest="table")
})
# @async_helpers.async_suppress
@asyncio.coroutine
async def predictaa(params):
    # result = Result.get_model(user_id=params["user_id"], model_id=params["model_id"])
    result = Result.get_model(**params)
    model_id_value = params['model_id']
    if result is None:
        raise OperatorException("error", f"模型{model_id_value}不存在！")
    parameters = json.loads(result.parameterJson)  # 从请求体中获取要获取的数据。
    model_class = parameters.pop("model_class")
    parameters["user_id"] = params['user_id']
    parameters["process_id"] = result.targetId
    parameters["output_port_id"] = 27
    # model = eval("%s(**parameters)" % model_class) 

    data = json.loads(request.data)
    features = data.get("features")
    model_type = data.get("model_type")
    model_params = data.get("model_params")
    data_set = data.get("data")

    if model is None:
        return jsonify({'error': 'Model not found'})
    try:
    #     loop = asyncio.get_event_loop()
    #     result = await loop.run_in_executor(None, model.transform(data_set, 'prediction'))
    #     # result = await model.transform(data_set, 'prediction')
    #     response = result.to_dict(orient='records')
    #     return jsonify(response)
    # except Exception as e:
    #     return jsonify({'error': str(e)})
        result = await model.transform(data_set, 'prediction')
        response = result.to_dict(orient='records')
        return jsonify(response)
    except Exception as e:
        return jsonify({'error': str(e)})
    # # predictions, _ = await op.do_work(model, data_set)
    # prediction_coroutine, _ = op.do_work(model, data_set)
    # async with prediction_coroutine as prediction:
    #     prediction_dict = prediction.to_dict()
    #     return jsonify(prediction_dict)
    # result = predictions.to_dict(orient="records")
    # return jsonify(result)

class LinearSVCModel(Model):
    def __init__(self, user_id, process_id, output_port_id, features, label, label_type, neg, pos, **kwargs):
        self.label = label
        self.label_type = label_type
        self.neg = neg
        self.pos = pos
        self.coefficients = kwargs["coefficients"]
        self.intercept = kwargs["intercept"]
        super(LinearSVCModel, self).__init__(user_id, process_id, output_port_id, features)

    async def transform(self, unl, table_name):
        lab = await super(LinearSVCModel, self).transform(unl, table_name)
        prediction = "`prediction_%s`" % self.label[1:-1]
        linear_item_sql_list = ["`%s` * (%f)" % (self.features[i], self.coefficients[i]) for i in
                                range(len(self.features))]
        linear_sql = "SELECT %s + %f AS hr_temp_col_linear, * FROM %s" % (
            " + ".join(linear_item_sql_list), self.intercept, lab.hr_view)

        lab.loc[prediction] = [self.label_type, "prediction",
                               "CASE WHEN hr_temp_col_linear >= 0 THEN '%s' ELSE '%s' END" % (
                                   self.pos, self.neg)]
        lab = lab.hr_sort_by_role()
        await lab.hr_update_view(table_name, lab.hr_get_view_sql(from_="(%s) hr_temp_table_lsvc_model" % linear_sql))
        return lab

解决建议 #

方案1：切换到ASGI服务器启用原生异步支持 #

Flask 2.0+原生支持异步视图，但默认WSGI服务器无法处理异步函数，需使用ASGI服务器运行应用：

1. 安装ASGI服务器（以uvicorn为例）：

pip install uvicorn

2. 创建asgi.py文件作为启动入口：

from your_app_module import app

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

3. 启动服务：

uvicorn asgi:app --reload

4. 调整视图代码：移除@asyncio.coroutine装饰器（该装饰器为旧版协程语法，与async def冲突），保留async def predictaa(params)的定义即可。

方案2：在WSGI模式下将异步函数转为同步执行 #

如果无法切换ASGI服务器，可使用asgiref工具将异步逻辑转为WSGI兼容的同步函数：

1. 安装依赖：

pip install asgiref

2. 修改视图代码：

from asgiref.sync import async_to_sync

@bp.route("/predict", methods=["POST"])
@pre.catch({...})
def predictaa(params):
    # 封装异步逻辑
    async def async_predict():
        result = Result.get_model(**params)
        model_id_value = params['model_id']
        if result is None:
            raise OperatorException("error", f"模型{model_id_value}不存在！")
        parameters = json.loads(result.parameterJson)
        model_class = parameters.pop("model_class")
        parameters["user_id"] = params['user_id']
        parameters["process_id"] = result.targetId
        parameters["output_port_id"] = 27
        # 恢复模型实例化代码，原代码中此行为注释状态
        model = eval(f"{model_class}(**parameters)")

        data = json.loads(request.data)
        data_set = data.get("data")

        if model is None:
            return jsonify({'error': 'Model not found'})
        try:
            result = await model.transform(data_set, 'prediction')
            response = result.to_dict(orient='records')
            return jsonify(response)
        except Exception as e:
            return jsonify({'error': str(e)})
    
    # 转换为同步执行并返回结果
    return async_to_sync(async_predict)()

方案3：用线程池执行异步任务 #

如果model.transform是IO密集型操作，可通过run_in_executor在同步视图中执行异步任务：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@bp.route("/predict", methods=["POST"])
@pre.catch({...})
def predictaa(params):
    result = Result.get_model(**params)
    model_id_value = params['model_id']
    if result is None:
        raise OperatorException("error", f"模型{model_id_value}不存在！")
    parameters = json.loads(result.parameterJson)
    model_class = parameters.pop("model_class")
    parameters["user_id"] = params['user_id']
    parameters["process_id"] = result.targetId
    parameters["output_port_id"] = 27
    model = eval(f"{model_class}(**parameters)")

    data = json.loads(request.data)
    data_set = data.get("data")

    if model is None:
        return jsonify({'error': 'Model not found'})
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        # 提交异步任务到线程池执行
        result = loop.run_until_complete(model.transform(data_set, 'prediction'))
        response = result.to_dict(orient='records')
        return jsonify(response)
    except Exception as e:
        return jsonify({'error': str(e)})

关键注意事项 #

• 必须移除@asyncio.coroutine装饰器：该装饰器用于旧版生成器协程，与async def原生协程冲突，会导致视图返回协程对象而非有效响应。
• 恢复模型实例化代码：原代码中model = eval(...)被注释，会导致model变量未定义，无法调用model.transform方法。
• Flask版本适配：Flask 2.0+的异步支持需要ASGI服务器配合，WSGI模式下必须通过转换工具适配异步逻辑。

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