flask使用async await处理异步任务

Question

class generatePs(Resource):
    async def post(self):
        data1 = request.get_data()
        data = json.loads(data1)
        uuid = data.get('uuid')
        current_id = data.get('current_id')
        tasks = []
        try:
            mt_obj = Outline.query.filter(Outline.uuid == uuid,
                                          Outline.current_name == current_id).first()

            task = asyncio.create_task(mt_deal(mt_obj))
            tasks.append(task)
            #
            task = asyncio.create_task(ib_deal(mt_obj))
            tasks.append(task)

            task = asyncio.create_task(ea_deal(mt_obj))
            tasks.append(task)

            task = asyncio.create_task(rp_deal(mt_obj))
            tasks.append(task)

            task = asyncio.create_task(generateAb(mt_obj))
            tasks.append(task)

            await asyncio.wait(tasks,timeout=None)

            return {'code': 200, 'message': 'success'}
        except Exception as e:
            return {'code': 400, 'message': str(e)}

async def ib_deal(obj):
    ibcontent_experience = eval(obj.ib)['ibexperience']
    ibcontent = eval(obj.ib)['ibcontent']
    iblinear = eval(obj.ib)['iblinear']
    linear_content = "You are encouraged to use a non-linear narrative to make your story focus on highlighting the skills and lessons of %s learned from the whole career and how they have shaped your growth." % ibcontent
    prompt = ""
    for experience in ibcontent_experience:
        if experience['label'] == '1':
                prompt += "As a student, please use following materials to write a within 200 words statement of your work experience. To make your statement compelling, use ornate and embellished language. Begin your statement with a simple sentence or phrase that declares your motivation for your professional involvements. There is a specific important rule you must follow: DO NOT mention the name of positions more than once in the statement. \n For the %s, you should not just simply follow the above material but expand with more details including providing the following information autonomously: Your responsibilities; The challenges you faced and how you overcame them; Methodologies you adapted to complete your duties; What you learned and gained in terms of skill and knowledge." % experience.get(
                    "content", "")
        else:
                prompt += "As a student, please use following materials to write a within 200 words statement of your work experience. To make your statement compelling, use ornate and embellished language. Begin your statement with a simple sentence or phrase that declares your motivation for your professional involvements. There is a specific important rule you must follow: DO NOT mention the name of positions more than once in the statement. \n For the %s, write in a sense of paraphrase and reorganize the above materials in the way of STAR" % experience.get(
                    "content", "")
        if iblinear == '1':
            prompt += linear_content
        dataload = {'prompt': prompt}
        ib_result = await requests.post(openai_url, json=dataload, headers=headers)
        obj.ibgpt = ib_result.text
        db.session.commit()

运行报错

    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type coroutine is not JSON serializable

Answer 1

【以下回答由 GPT 生成】

优化后的代码如下所示：

from flask import Flask, request, jsonify
from flask_restful import Resource, Api
import asyncio
import json

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


class GeneratePs(Resource):
    async def post(self):
        data = json.loads(request.get_data())
        uuid = data.get('uuid')
        current_id = data.get('current_id')
        tasks = []
        try:
            mt_obj = Outline.query.filter(Outline.uuid == uuid, Outline.current_name == current_id).first()

            tasks.append(mt_deal(mt_obj))
            tasks.append(ib_deal(mt_obj))
            tasks.append(ea_deal(mt_obj))
            tasks.append(rp_deal(mt_obj))
            tasks.append(generateAb(mt_obj))

            await asyncio.gather(*tasks)

            return {'code': 200, 'message': 'success'}
        except Exception as e:
            return {'code': 400, 'message': str(e)}


async def mt_deal(obj):
    # Perform some async tasks related to mt_obj
    # ...


async def ib_deal(obj):
    # Perform some async tasks related to ib_obj
    # ...


async def ea_deal(obj):
    # Perform some async tasks related to ea_obj
    # ...


async def rp_deal(obj):
    # Perform some async tasks related to rp_obj
    # ...


async def generateAb(obj):
    # Perform some async tasks related to generateAb_obj
    # ...


api.add_resource(GeneratePs, '/generate_ps')

if __name__ == '__main__':
    app.run()

解释：

1. 创建GeneratePs类继承自Resource类，用于处理POST请求。
2. 在post方法中，首先解析请求中的数据，然后创建一个空的任务列表tasks。
3. 通过查询数据库获取mt_obj对象。
4. 将每个异步任务添加到任务列表中，使用append方法。这里假设mt_deal、ib_deal、ea_deal、rp_deal和generateAb函数都是异步的。
5. 使用asyncio.gather方法来并发地运行所有异步任务，并等待它们全部完成。
6. 如果出现异常，捕获并返回错误信息。
7. 如果没有异常，返回成功的响应。
8. 创建每个异步任务的函数mt_deal、ib_deal、ea_deal、rp_deal和generateAb，根据需要执行与任务相关的异步操作。

这样，我们通过使用async和await关键字，以及asyncio库中的方法，可以在Flask中处理异步任务。

如果你已经解决了该问题, 非常希望你能够分享一下解决方案, 写成博客, 将相关链接放在评论区, 以帮助更多的人 ^-^