Celery sending tasks as a group

Question

So I have some tasks in celery that are not registered in the current process and what I am trying to achieve is to actually make a group call for multiple tasks. For example:

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

The problem is that using send_task is group will give me this error

Traceback (most recent call last):
  File "workflows.py", line 17, in 
    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'

More of this, I tried using signature method to make calls but I don't know how to mock this signature during unit tests when testing on local env.

Any ideas ?

Celery sending tasks as a group

0 Answers