So geben Sie Argumente von Sellerie-Jobs in die abzufragende Datenbank ein; mysql verwenden

Question

Ich benutze MySql als Ergebnis Backend für Sellerie. Ich möchte auch Programmargumente in der DB speichern.

Zum Beispiel:

add.apply_async(args=[num1, num2, user] 

In diesem Fall habe ich Benutzer Argument in DB gespeichert werden sollen, so dass ich später abfragen.

Derzeit gebe ich Argumente zur Verfügung gestellt, die in der DB gespeichert ist.

def add(num1, num2, user): 
    return [num1+num2, user] 

Wenn der Task jedoch ausgeführt wird, wird der Benutzer nicht eingefügt, und ich kann ihn nicht in der DB abfragen. Gibt es dafür eine Lösung/einen Hack?

Answer 1

Verwenden Sie MySql nur als Ergebnis-Backend oder verwenden Sie es auch für die Warteschlange (was ich nicht empfehlen würde)? Wenn Sie es für eine Warteschlange verwenden, sollten sich die Argumente in der Datenbank befinden, sobald die Aufgabe gesendet wurde. Andernfalls kann das Task-Ergebnis erst gespeichert werden, nachdem die Task beendet wurde.

Wenn die Argumente während der Ausführung der Aufgabe abfragbar sein sollen, müssen Sie sie zu Beginn Ihrer Aufgabe manuell in eine DB-Tabelle einfügen. Wenn Sie möchten, dass sie vor dem Start der Task abfragbar sind, müssen Sie sie direkt vor dem Aufruf von apply_async in eine DB-Tabelle einfügen.

Answer 2

Angenommen, Sie können verbinden und wählen Sie die db, Sie möchten möglicherweise eine Funktion, die Parameter abfragen und eine Funktion callback trennen, um Ihre Ergebnisse in db zu speichern.

Add() werden Ergebnisse arbeiten, und store_callback() wird sie in db speichern, wenn sie fertig sind. Wenn also die Parameter bereit sind, kann Ihr Code mit der nächsten Aufgabe fortfahren und das Ergebnis (Benutzer) speichern, wenn Sie fertig sind.

Etwas wie:

def store_callback(result): 
    sql_insert = 'INSERT INTO your_table VALUES(?, ?,)' 
    curs.execute(sql_insert, result) #result is a list passed from add() 

def add(num1, num2, user): 
    return [num1+num2, user] 


# check parameters are present in db: 
curs.execute("SELECT * FROM your_table WHERE user = ?", [_user]) 
user_exists = curs.fetchone() 
# 
if user_exists: 
    add.apply_async((num1, num2, user,) , link=store_callback.s()) 

Sie selbst() auf eine andere Aufgabe Link hinzufügen können.

Answer 3

Sie benötigen Warteschlange, jeder Benutzer eine Warteschlange für sich selbst, Sellerie Aufgaben erhalten Argumente aus der Benutzerwarteschlange.

queue.get(timeout=10)# if nothing got then retry util get the arguments