Jak wprowadzać argumenty zadań selerowych do bazy danych, które mają być sprawdzane; używając mysql

Question

Używam MySql jako backend wyniku dla selera. Chcę również przechowywać argumenty programu w DB.

Na przykład:

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

W tym przypadku chcę, aby przechowywać argumentu użytkownika w DB, tak, że mogę zapytać później.

Obecnie zwracam podane argumenty, które są przechowywane w DB.

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

Jednak gdy zadanie jest uruchomione, użytkownik nie jest wstawiany i nie mogę go wysłać do DB. Czy istnieje jakieś rozwiązanie/hack do tego?

Answer 1

Czy używasz MySql jako zaplecza wyniku, czy też używasz go również do kolejki (której nie polecałbym)? Jeśli używasz go do kolejki, argumenty powinny znajdować się w bazie danych zaraz po wysłaniu zadania. W przeciwnym razie wynik zadania nie może zostać zapisany, dopóki zadanie nie zostanie zakończone.

Jeśli potrzebujesz argumentów, aby można było wywoływać zapytania podczas wykonywania zadania, musisz ręcznie wstawić je do tabeli DB na początku zadania. Jeśli chcesz, aby były możliwe kwerendy przed rozpoczęciem zadania, musisz je wstawić do tabeli DB tuż przed wywołaniem apply_async.

Answer 2

Zakładając, że możesz połączyć się i wybrać do bazy danych, , możesz chcieć oddzielić funkcję, której parametry zapytania i funkcję callback, aby zapisać wyniki w db.

Add() będzie działać, a store_callback() zapisze je w db, gdy będzie gotowe. Więc jeśli parametry są gotowe, twój kod może przejść do następnego zadania i zapisać wynik (użytkownika) po zakończeniu.

Coś jak:

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

można nawet link Dodaj() do innego zadania.

Answer 3

Potrzebujesz kolejki, każdy użytkownik buduje kolejkę dla siebie, zadania selera pobierają argumenty z kolejki użytkownika.

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