Celery - 任务签名转换为字典，无法调用和弦 - 在和弦的标题组中使用任务链答案

【问题标题】：Celery - Task signature converted to dictionary and cannot call in chord - Using chain of tasks in header group of chordCelery - 任务签名转换为字典，无法调用和弦 - 在和弦的标题组中使用任务链
【发布时间】：2020-03-23 22:36:00
【问题描述】：

关注celery chord with group of chains with exception in chain的答案

我收到一个错误，其中 Celery 似乎将任务签名链更改为导致以下错误的字典。

芹菜 - 4.4 Redis - 3.3.11

代码 - 来自@bigzbig

@celery_app.task
def task_one():
    return 'OKIDOKI'

@celery_app.task
def task_two(str):
    return f'{str} YOUPI'

@celery_app.task
def task_three(str):
    return f'{str} MAKAPAKA'

@celery_app.task
def task_exception(str):
    raise KeyError(f'{str} Ups')

@celery_app.task(ignore_result=True)
def task_wrapper(*args, **kwargs):
    if 'job' in kwargs:
        kwargs['job'].apply()

@celery_app.task(ignore_result=True)
def callback_task(*args, **kwargs):
    return (args, kwargs, 'Yeah')

def test():
    chains = []

    tasks = [
        task_one.s(),
        task_two.s(),
        task_exception.s(),
        task_three.s(),
    ]
    chains.append(task_wrapper.s(job=chain(*tasks)))

    tasks = [
        task_one.s(),
        task_two.s(),
        task_three.s(),
    ]
    chains.append(task_wrapper.s(job=chain(*tasks)))

    chord(chains, callback_task.s()).apply_async()

kwargs['job']的打印

celeryworker2_1  | [2020-03-23 22:31:01,646: WARNING/ForkPoolWorker-1] {'task': 'celery.chain', 'args': [], 'kwargs': {'tasks': [{'task': 'portfolio.tasks.task_one', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'chord_size': None, 'immutable': False}, {'task': 'portfolio.tasks.task_two', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'chord_size': None, 'immutable': False}, {'task': 'portfolio.tasks.task_exception', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'chord_size': None, 'immutable': False}, {'task': 'portfolio.tasks.task_three', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'immutable': False, 'chord_size': None}]}, 'options': {}, 'subtask_type': 'chain', 'immutable': False, 'chord_size': None}
celeryworker_1   | [2020-03-23 22:31:01,650: WARNING/ForkPoolWorker-1] {'task': 'celery.chain', 'args': [], 'kwargs': {'tasks': [{'task': 'portfolio.tasks.task_one', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'chord_size': None, 'immutable': False}, {'task': 'portfolio.tasks.task_two', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'chord_size': None, 'immutable': False}, {'task': 'portfolio.tasks.task_three', 'args': [], 'kwargs': {}, 'options': {}, 'subtask_type': None, 'immutable': False, 'chord_size': None}]}, 'options': {}, 'subtask_type': 'chain', 'immutable': False, 'chord_size': None}

错误

Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 385, in trace_task
R = retval = fun(*args, **kwargs)
File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 650, in __protected_call__
return self.run(*args, **kwargs)
File "/app/portfolio/tasks.py", line 241, in task_wrapper
kwargs['job'].apply()
AttributeError: 'dict' object has no attribute 'apply'

【问题讨论】：

您是否尝试使用某些旧 Celery 版本（4.2.x 或更早版本）运行相同的代码？
刚用 Celery 4.2 试过，但同样的问题，我会再试几次再回来。
还要查看 Celery 问题和 PR - 也许问题已被记录和/或修复。

标签： task celery chain chord

【解决方案1】：

您正试图将签名传递给另一个任务。所以 Celery 将其转换为 dict。您可以从 dict 构建签名。

"task_always_eager=True" 设置为在同一进程下运行，而不是作为不同的 celery 任务运行，因为执行链本身是不同的任务。这样，您将保留链接或 link_error 是任何给定的。

from celery.canvas import Signature
callback = Signature(kwargs['job'])
callback.delay(task_always_eager=True)

【讨论】：