Django Sellerie Arbeiter zu Echtzeit-Status-und Ergebnisnachrichten an Front-End senden

Question

In einer Django-App bin ich async Aufgaben laufen und möchte dem Benutzer Fortschritt, Fehler usw. anzeigen. Wenn Fehler auftreten, sollte der Benutzer auf eine Seite umgeleitet werden, auf der zusätzliche Eingaben oder Aktionen erforderlich sind, um das Problem zu beheben. Wie kommuniziert man am besten von der Sellerie zurück zum Frontend?

Hier ist eine Grundstruktur in Pseudo-Code:

# 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} 

Was würde ich im Idealfall wollen:

• compute_module.py Modul Python nativen Logger verwendet. Durch die Trennung der Aufgaben möchte ich das Logging so allgemein wie möglich halten und die Standard-Python/Django-Logger verwenden. Aber sie scheinen nicht dafür ausgelegt zu sein, Nachrichten an das Frontend zu senden.
• Sellerie Aufgabe behandelt irgendwie die Protokolle und anstatt sie auf stdout Anzeige leitet sie
• Front-End-js zeigt und bearbeitet die Meldungen der Kommunikation zwischen Sellerie Arbeiter und Front

Es könnte Standard Möglichkeiten, um Drücker Ende, dass ich nicht bewusst bin. Dieses Szenario muss oft passieren und ich bin überrascht, dass es so schwer zu implementieren ist. in gewisser Weise sollte die Rabbitmq Message Queue oder aws sns dafür ausgelegt sein. Unten sind Ressourcen, die ich angeschaut habe, aber nicht beide gut funktionieren, aber vielleicht bin ich nur verwirrt.

Protokollierung: dies scheint über das Anmelden auf der Server-Seite mehr zu sein, keine Nachrichten an Benutzer

• 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

Sellerie cam Senden scheint über Admin-Überwachungsaufgaben zu sein, nicht sen ding-Nachrichten an Benutzer

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

Schieber Ich mag aber ich will nicht compute_module.py Deal mit ihm haben. Zum Beispiel möchte ich keine pusher.com-Integration in compute_module.py machen. Ich glaube, ich könnte einen Schieber-Objekt übergeben, die bereits instanziiert wurde so das Modul nur Push-Nachrichten können aber auch hier würde ich es vorziehen, generische

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

Answer 1

Der einzige Weg, ich zu sein‘ Es ist mir gelungen, Realtime-Status zu erhalten, indem ich einfach einige SQL-Write/API-Aufrufe in die Task selbst lege. Dinge mit dem Rückgabewert der Aufgabe zu tun ist viel einfacher, da Sie einfach eine benutzerdefinierte Aufgabenklasse schreiben können.

Ich bin nicht ganz sicher, wie das mit Django funktioniert, aber es sollte in etwa so aussehen. http://docs.celeryproject.org/en/latest/userguide/tasks.html#handlers

Answer 2

Ok, so unten ist Pseudo-Code für wie ich gelöst habe es jetzt:

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() 

Die vollständige Liste finden Sie hier. Grundsätzlich verwende ich https://pusher.com/docs/javascript_quick_start und serverseitig übergeben Sie das instanziierte Objekt in die compute_module. Ein Nachteil ist, dass die Pusher-Nachrichten sind ephermeral, also werde ich etwas mehr Arbeit in LogPusher tun müssen, um sie in einem db zu speichern, etwas für einen anderen Tag ...

Auch in meiner realen Umsetzung trigger ich die Aufgabe über einen $.post() Ajax Anruf in $(document).ready(), da kleine Aufgaben so schnell abgeschlossen werden, würde der Benutzer nie die Pusher-Nachrichten sehen, weil die Verbindung nicht hergestellt wurde (zurück zu diesem historischen Nachrichtenproblem).

Eine weitere Alternative Route, die ich oben nicht erwähnt hatte, ist https://channels.readthedocs.io/en/latest/

# views.py 
from tasks import run_task 

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


# tasks.py 
import pusher 
from django.conf import settings 
from compute_module import compute_fct 

class LogPusher(object): 
    def __init__(self, event): 
     self.pusher_client = pusher.Pusher(app_id=settings.PUSHER_APP_ID, 
         key=settings.PUSHER_KEY, 
         secret=settings.PUSHER_SECRET, 
         cluster=settings.PUSHER_CLUSTER, ssl=True) 
     self.event = event 

    def send(self, data): 
     self.pusher_client.trigger(settings.PUSHER_CHANNEL, self.event, json.dumps(data)) 

@shared_task 
def run_task(pusher_event): 

    log_pusher = LogPusher(pusher_event) 
    result = compute_fct(log_pusher) 

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

    if result == 'error': 
      log_pusher.send('status':'error') 
    else: 
      log_pusher.send('status':'success') 


# compute_module.py 
import pandas as pd 

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

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


# context_processors.py 
# see https://stackoverflow.com/questions/433162/can-i-access-constants-in-settings-py-from-templates-in-django 
from django.conf import settings 

def pusher(request): 
    return {'PUSHER_KEY': settings.PUSHER_KEY, 'PUSHER_CLUSTER': settings.PUSHER_CLUSTER , 'PUSHER_CHANNEL': settings.PUSHER_CHANNEL } 


# template.html 
<script> 

var pusher = new Pusher("{{PUSHER_KEY}}", { 
    cluster: "{{PUSHER_CLUSTER}}", 
    encrypted: true  
}); 

var channel = pusher.subscribe("{{PUSHER_CHANNEL}}"); 
channel.bind("{{pusher_event}}", function(data) { 
    // process data 
}); 

</script>