具有进度跟踪的后台任务，没有芹菜或 Redis

Question

我的项目从一个服务导出到另一个服务。为此，需要一个较长的后台任务，并将进度显示为一些文本，例如进度current step。一个问题是如何从长任务中获取此文本。

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

我试图将这个current step text保存到session.但是long task期间的反应总是null.我认为是因为flask忙于任务并且没有返回session的有效值.已尝试在新线程中运行任务。然后我收到有关从错误线程访问会话的错误。

Answer 1

我的解决方案

创建全局字典

step = dict()

创建随机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']]}

现在，对于每个要导出的请求，将仅为其创建新的密钥和文本

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

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

del self.step[self.key]