Собственные параметры метода PeriodicTask run() в Celery

Question

Я пишу небольшое приложение Django, и я должен был бы создать для каждого объекта модели, его периодическая задача, которая будет выполнена с определенным интервалом. Я использовать для этого приложения сельдерея, но я не могу понять одну вещь:

class ProcessQueryTask(PeriodicTask): 
    run_every = timedelta(minutes=1) 

    def run(self, query_task_pk, **kwargs): 
     logging.info('Process celery task for QueryTask %d' % 
query_task_pk) 
     task = QueryTask.objects.get(pk=query_task_pk) 
     task.exec_task() 
     return True 

Тогда я буду делать следующее:

>>> from tasks.tasks import ProcessQueryTask 
>>> result1 = ProcessQueryTask.delay(query_task_pk=1) 
>>> result2 = ProcessQueryTask.delay(query_task_pk=2) 

Первый звонок успех, но и другие периодические звонки возвращаются ошибка - TypeError: run() принимает ровно 2 аргумента без ключевого слова (1 задано) в сервер celeryd. Могу ли я передать собственные параметры PeriodicTask run()?

Answer 1

Это был ответ замечательный Ask Solem в his response to your question на celery-users Google group.

Периодические задания не используют аргументы, поэтому вам нужно сделать несколько классов или выполнить периодическую задачу, которая обрабатывает более одной «модели».

т.д .:

from celery.task import PeriodicTask 
from celery.decorators import periodic_task 

# base class 
class BaseProcessQueryTask(PeriodicTask): 
    abstract = True 
    run_every = timedelta(minutes=1) 
    query_task_pk = None 

    def run(self): 
     task = QueryTask.objects.get(pk=self.query_task_pk) 
     task.exec_task() 

class ProcessQueryTask1(BaseProcessQueryTask): 
    query_task_pk = 1 

class ProcessQueryTask2(BaseProcessQueryTask): 
    query_task_pk = 2 

но это более вероятно, что вы хотите что-то вроде этого:

@task(ignore_result=True) 
def execute_query_task(task): 
    task.exec_task() 

@periodic_task(run_every=timedelta(minutes=1)) 
def process_query_tasks(): 
    for task in QueryTask.objects.all(): 
     ExecuteQueryTask.delay(task)