Django芹菜工作者向前端发送实时状态和结果消息

Question

在django应用程序中,我正在运行异步任务,并希望向用户显示进度,错误等.如果存在错误,则应将用户重定向到需要其他输入或某些操作来解决问题的页面.从芹菜工作到前端沟通的最佳方式是什么？

这是伪代码的基本结构:

# views.py
from tasks import run_task

def view_task():
    run_task.delay()
    return render(request, 'template.html')

# tasks.py
from compute_module import compute_fct

@shared_task
def run_task():
    result = compute_fct()

    # how to catch status update messages from compute_module while compute_fct is running??

    if result == 'error':
        handle_error()
    else:
        handle_succes()     

# compute_module
import pandas as pd

def compute_fct():
    # send message: status = loading file
    df = pd.read_csv('test.csv')
    # send message: status = computing
    val = df['col'].mean()

    if val is None:
        return {'status':'error'}
    else:
        return {'status':'success','val':val}

我理想的想要的是:

• compute_module.py模块使用python本机记录器.通过职责分离,我希望尽可能保持日志记录的通用性,并使用标准的python/django记录器.但它们似乎并不是为了向前端发送消息而设计的.
• celery任务以某种方式处理日志而不是在stdout上显示它们将它们重定向到推动器
• 前端js显示和处理消息

芹菜工人和前端之间可能有标准的沟通方式,我不知道.这种情况必须经常发生,我很惊讶它实施起来很困难.在某种程度上,应该为此设计rabbitmq消息队列或aws sns.下面是我看过的资源,但感觉它们中的任何一个都不能很好地工作,但也许我只是感到困惑.

logging:这似乎更多的是关于登录服务器端,而不是向用户发送消息

• http://docs.celeryproject.org/en/latest/userguide/tasks.html#logging
• https://docs.djangoproject.com/en/2.0/topics/logging/
• http://oddbird.net/2017/04/17/async-notifications/
• https://www.google.com/search?q=celery+worker+send+message+to+front+end

Celery cam似乎是关于管理员监控任务,而不是向用户发送消息

• http://docs.celeryproject.org/en/latest/userguide/monitoring.html

推动我喜欢,但我不想compute_module.py处理它.那就是例如我不想在里面进行任何pusher.com集成compute_module.py.猜猜我可以传递一个已经实例化的推送器对象,这样模块可以只推送消息但是我希望它再次是通用的

• https://blog.pusher.com/improve-user-experience-app-real-time-progress-bar-tutorial/
• https://blog.pusher.com/django-pusherable/

Answer 1

我设法获取实时状态的唯一方法是简单地将一些 SQL 写入/API 调用放入任务本身。使用任务的返回值执行操作要容易得多，因为您只需编写自定义任务类即可。

我不完全确定使用 Django 是如何工作的，但它应该看起来像这样。

class CustomTask(celery.Task):
    def __call__(self, *args, **kwargs):
        self.start_time = time.time()

    def on_success(self, retval, task_id, args, kwargs):
        do_success_stuff()

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        do_failure_stuff()

@shared_task(base=CustomTask)
def do_stuff():
    return create_widgets()

完整列表可以在这里找到： http: //docs.celeryproject.org/en/latest/userguide/tasks.html#handlers