如何从Flask后台任务中获取含进度百分比的详细状态？

问题描述 #

需要在Flask API中实现耗时极长的异步任务，计划用flask_executor，但现有代码只能返回任务的"running"或"completed"状态，想要在slow_task_status接口中获取任务的完成百分比等详细信息。

原代码如下：

from flask import Flask, request, jsonify
from flask_executor import Executor
import uuid
import time
 
app = Flask(__name__)
executor = Executor(app)
 
 
class BackgroundRunner:
    def __init__(self, executor):
        self.executor = executor
 
    def task(self):
        percent_completed = 0
        time.sleep(1000)
        percent_completed = 33
        time.sleep(1000)
        percent_completed = 66
        time.sleep(1000)
        percent_completed = 100
 
    def task_async(self):
        task_id = uuid.uuid4().hex # generate a unique task ID
        task = self.executor.submit(task_id)
        return task_id
 
    def task_status(self, task_id):
        task = self.executor.futures._state(task_id)
        if task.done():
            return "completed"
        else:
            return "running"
 
background_runner = BackgroundRunner(executor)
 
 
@app.route('/slow_task', methods=['POST'])
def slow_task():
    task_id = background_runner.task_async()
    return jsonify({"task_id": task_id})
 
 
@app.route('/slow_task_status/<task_id>')
def slow_task_status(task_id):
    status = background_runner.task_status(task_id)
    return jsonify({"status": status})
 
 
if __name__ == '__main__':
    app.run()

解决方案 #

Flask-Executor本身没有内置进度跟踪功能，但可以通过维护一个任务进度存储（比如内存字典）来实现，核心思路是在异步任务执行过程中主动更新进度，然后在状态接口中读取该进度。

修改后的完整代码：

from flask import Flask, request, jsonify
from flask_executor import Executor
import uuid
import time
 
app = Flask(__name__)
executor = Executor(app)
 
 
class BackgroundRunner:
    def __init__(self, executor):
        self.executor = executor
        # 新增字典存储每个任务的进度，key为task_id，value为百分比
        self.task_progress = {}
 
    def task(self, task_id):
        # 初始化进度为0
        self.task_progress[task_id] = 0
        time.sleep(10)  # 替换为实际的耗时操作
        self.task_progress[task_id] = 33
        time.sleep(10)
        self.task_progress[task_id] = 66
        time.sleep(10)
        self.task_progress[task_id] = 100
 
    def task_async(self):
        task_id = uuid.uuid4().hex
        # 提交任务时传入task_id，让任务能更新对应进度
        self.executor.submit(self.task, task_id)
        return task_id
 
    def task_status(self, task_id):
        # 先检查任务是否存在
        if task_id not in self.task_progress:
            return {"status": "not_found", "progress": 0}
        
        # 获取任务状态
        future = self.executor.futures.get(task_id)
        if future and future.done():
            status = "completed"
        else:
            status = "running"
        
        # 返回状态和当前进度
        return {
            "status": status,
            "progress": self.task_progress[task_id]
        }
 
background_runner = BackgroundRunner(executor)
 
 
@app.route('/slow_task', methods=['POST'])
def slow_task():
    task_id = background_runner.task_async()
    return jsonify({"task_id": task_id})
 
 
@app.route('/slow_task_status/<task_id>')
def slow_task_status(task_id):
    result = background_runner.task_status(task_id)
    return jsonify(result)
 
 
if __name__ == '__main__':
    app.run(debug=True)

关键修改点说明 #

• 新增self.task_progress字典：用来存储每个任务的实时进度，键为任务ID，值为完成百分比。
• 修改task方法：接收task_id参数，在每个阶段更新对应任务的进度值。
• 调整task_async：提交任务时传入task_id，确保任务能找到对应的进度存储项。
• 升级task_status：不仅返回任务状态，还从task_progress中读取当前进度，同时处理任务不存在的情况。

注意事项 #

• 如果你的Flask应用使用多进程模式（比如app.run(processes=2)），内存字典task_progress无法在进程间共享，此时需要改用外部存储（如Redis、数据库）来保存进度。
• 任务完成后可以选择清理task_progress中的旧数据，避免内存占用过高。

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