没有 Celery 或 Redis 的进度跟踪后台任务

Question

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

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

我试图将此 current step text 保存到 session 中。但是 long task 期间的响应总是 null。我认为是因为 flask 正忙于任务并且 return 不是 session 的有效值。试图运行新线程中的任务。然后我收到有关从错误线程访问会话的错误。

Answer 1

我的解决方案

创建全局字典

step = dict()

随机创建 key 并通过全局步骤将此密钥赋予模块后台任务。开始一个新线程和 return 键。

@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]

没有 Celery 或 Redis 的进度跟踪后台任务

Background task with progress tracking without Celery or Redis

python

redis

celery

python-multithreading

flask

我的解决方案