python - 无需 Celery 或 Redis 即可进行进度跟踪的后台任务

Question

我的项目做了一些从一项服务到另一项服务的导出。为此，需要一个很长的后台任务，并将进度显示为一些文本，例如进度的当前步骤。一个问题是如何从长任务中获取这段文本。

我了解 Celery 和 Redis。但它需要额外的资源，例如服务器。我的项目太小，每月参加的人数不会超过几个人。所以我不想购买共享主机或机器。

我尝试将此当前步骤文本保存到 session 中。但长任务期间的响应始终为null。我认为是因为 Flask 正忙于任务并且没有为 session 返回有效值。尝试在新线程中运行该任务。然后我收到有关从错误线程访问 session 的错误。

Answer 1

我的解决方案

创建全局字典

step = dict()

创建随机 key ，并为该 key 提供全局步骤进入模块的后台任务。启动一个新线程并返回键。

@app.route('/api/transfer', methods=['post'])
def api_transfer():
    global step
    key = token_urlsafe(10)

    service = Service(step, key, session.get('token'))

    t = threading.Thread(target=long_task, args=(service,))
    t.start()

    return {'key': key}

并使用此键发送请求以获取其当前步骤的响应

@app.route('/api/transfer/current_step', methods=['post'])
def api_current_step():
    global step
    return {'step': step[request.json['key']]}

现在，对于每个导出请求，都会为其创建新的 key 和文本

{'jlIG7VKtbMHtXg': 'some step for jlIG7VKtbMHtXg'}
{'jlIG7VKtbMHtXg': 'some step for jlIG7VKtbMHtXg', 'wxr4jKyP7c72sg': 'some step for wxr4jKyP7c72sg'}

并在任务结束时，从字典中删除此键

del self.step[self.key]