python - Celery 群发任务

Question

所以我在 celery 中有一些任务没有在当前进程中注册，而我想要实现的是实际为多个任务进行组调用。例如：

from celery import group, Signature, signature
from task_app import celery_app
from celery.execute import send_task

workflow = group([
    send_task(
        'my_task.deploy',
        kwargs={
            'name': 'dev1',
            'site': 'somesite',
            'sleep': 1
        },
        immutable=True,
        app=celery_app,
    )]
)
result = workflow()
print(result.get())

问题是使用send_taskis group 会给我这个错误

Traceback (most recent call last):
  File "workflows.py", line 17, in <module>
    result = workflow()
  File "/home/env/lib/python3.8/site-packages/celery/canvas.py", line 1077, in __call__
    return self.apply_async(partial_args, **options)
  File "/home/env/lib/python3.8/site-packages/celery/canvas.py", line 1102, in apply_async
    results = list(self._apply_tasks(tasks, producer, app, p,
  File "/home/env/lib/python3.8/site-packages/celery/canvas.py", line 1182, in _apply_tasks
    for sig, res in tasks:
  File "/home/env/lib/python3.8/site-packages/celery/canvas.py", line 1163, in _prepared
    task = from_dict(task, app=app)
  File "/home/env/lib/python3.8/site-packages/celery/canvas.py", line 139, in from_dict
    typ = d.get('subtask_type')
  File "/home/env/lib/python3.8/site-packages/celery/result.py", line 223, in get
    return self.backend.wait_for_pending(
  File "/home/env/lib/python3.8/site-packages/celery/backends/asynchronous.py", line 199, in wait_for_pending
    for _ in self._wait_for_pending(result, **kwargs):
  File "/home/env/lib/python3.8/site-packages/celery/backends/asynchronous.py", line 265, in _wait_for_pending
    for _ in self.drain_events_until(
  File "/home/env/lib/python3.8/site-packages/celery/backends/asynchronous.py", line 51, in drain_events_until
    if timeout and time.monotonic() - time_start >= timeout:
TypeError: '>=' not supported between instances of 'float' and 'str'

更多的是，我尝试使用signature方法进行调用，但我不知道如何在本地环境上进行单元测试时模拟这个签名。

有任何想法吗 ？

标签： pythondjangocelerypytest