Сельдерей обрабатывает именованный аргумент

У меня есть такая задача сельдерея

@app.task(bind=True,max_retries=3, default_retry_delay=1 * 60)  
def doTargetprefilter(self,*args,**kwargs ):

Я называю это как

args = [None,sourcedns, targetdnlist]
kwargs= {'workername':workername,}
result = r.doTargetprefilter.apply_async(*args,**kwargs)

Однако я получаю странную ошибку

File "/usr/local/lib/python3.4/dist-packages/celery/app/amqp.py", line 254, in publish_task
raise ValueError('task kwargs must be a dictionary')

ValueError: задача kwargs должна быть словарем

Небольшой модульный тест вызова работает нормально;

def test_doTargetprefilter(self):
    from  ltecpxx.mrosimpleexecutor import doTargetprefilter
    s=[1,2,3]
    t=[1,2,2]
    workername="ltecpxx.mrosimpleexecutor.SimplePrefilter"
    args =[None,s,t]
    kwargs={'wokername':workername}
    doTargetprefilter(*args,**kwargs)

Я пробовал всевозможные комбинации, а также видел документацию по apply_async. Это работает, если я делаю это как обычный метод (без *args и **kwargs); Что я делаю не так


python-3.x celery celery-task
person Alex Punnen    schedule 13.07.2015    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Аннотация bind предоставляет self; Поэтому нам нужно удалить это из списка аргументов, и все аргументы должны быть в кортеже, когда мы вызываем apply_async. Изменение этих двух даст

args = [sourcedns, targetdnlist]
kwargs= {'workername':workername}
result = r.doTargetprefilter.apply_async((args),kwargs)

И сигнатура функции

@app.task(bind=True,max_retries=3, default_retry_delay=1 * 60)  # retry in 1 minutes.
def doTargetprefilter(self,*args,workername=""):
person Alex Punnen    schedule 14.07.2015