Flask API报错TypeError:协程对象无法JSON序列化问题求助

你遇到的TypeError: Object of type coroutine is not JSON serializable本质是Flask/Flask-RESTful默认不支持异步装饰器直接装饰同步视图：

• 你的require_api_key_from_db装饰器返回的是异步wrapper函数，当Flask调用这个装饰后的视图方法时，得到的是一个coroutine对象，而非标准的响应元组/字典；
• 同时你在wrapper里用await func(*args, **kwargs)调用同步视图方法，会把同步函数包装成coroutine，最终返回的result也是coroutine，Flask尝试将其序列化为JSON时自然报错。

方案1：改用异步视图（推荐，Flask 2.0+支持） #

将Resource中的同步方法改为异步方法，利用Flask 2.0及以上版本的原生异步支持：

from flask import Flask, request
from functools import wraps
import asyncio
from flask import jsonify

from flask_restful import Resource, Api, reqparse
from database import DataBase

app = Flask(__name__)
api = Api(app)

database = DataBase()

data_storage = {}

def require_api_key_from_db(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        api_key = request.headers.get("X-API-Key")
        if api_key:
            user_id = await database.get_user_id_by_api_key(api_key)
            if user_id is None:
                return {"error": "Unauthorized 2"}, 401
            else:
                kwargs["user_id"] = user_id
                result = await func(*args, **kwargs)
                return result
        return {"error": "Unauthorized"}, 401

    return wrapper

class DataResource(Resource):
    method_decorators = [require_api_key_from_db]

    # 改为async方法
    async def get(self, key):
        if key in data_storage:
            return {key: data_storage[key]}
        else:
            return {"error": "Not Found"}, 404

    async def put(self, key):
        parser = reqparse.RequestParser()
        parser.add_argument("value", type=str, required=True, help="Value cannot be blank.")
        args = parser.parse_args()

        data_storage[key] = args["value"]
        return {key: data_storage[key]}

    async def delete(self, key):
        if key in data_storage:
            del data_storage[key]
            return {"result": "Deleted successfully"}
        else:
            return {"error": "Not Found"}, 404

api.add_resource(DataResource, "/data/<string:key>")

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

方案2：将异步操作同步化（兼容旧版Flask） #

如果无法升级Flask版本，可在装饰器中手动运行asyncio事件循环，把异步数据库调用转为同步执行：

from flask import Flask, request
from functools import wraps
import asyncio
from flask import jsonify

from flask_restful import Resource, Api, reqparse
from database import DataBase

app = Flask(__name__)
api = Api(app)

database = DataBase()

data_storage = {}

def require_api_key_from_db(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        api_key = request.headers.get("X-API-Key")
        if api_key:
            # 运行异步函数并获取结果，转为同步操作
            user_id = asyncio.run(database.get_user_id_by_api_key(api_key))
            if user_id is None:
                return {"error": "Unauthorized 2"}, 401
            else:
                kwargs["user_id"] = user_id
                # 直接调用同步视图方法，无需await
                result = func(*args, **kwargs)
                return result
        return {"error": "Unauthorized"}, 401

    return wrapper

class DataResource(Resource):
    method_decorators = [require_api_key_from_db]

    # 保持原同步方法不变
    def get(self, key):
        if key in data_storage:
            return {key: data_storage[key]}
        else:
            return {"error": "Not Found"}, 404

    def put(self, key):
        parser = reqparse.RequestParser()
        parser.add_argument("value", type=str, required=True, help="Value cannot be blank.")
        args = parser.parse_args()

        data_storage[key] = args["value"]
        return {key: data_storage[key]}

    def delete(self, key):
        if key in data_storage:
            del data_storage[key]
            return {"result": "Deleted successfully"}
        else:
            return {"error": "Not Found"}, 404

api.add_resource(DataResource, "/data/<string:key>")

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

• 方案1需确保Flask版本≥2.0，否则异步视图无法正常运行；
• 方案2中asyncio.run()会创建新事件循环，若你的数据库连接依赖全局事件循环，需调整DataBase类实现以适配；
• 两种方案都要保证database.get_user_id_by_api_key(api_key)返回可JSON序列化的类型（如字符串、整数），若返回自定义对象仍会触发序列化错误。

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