[Python] Correct type annotation for a celery task

I am looking for the correct type annotation to attribute to a celery task which has been set by the celery @task decorator.

Assume I have a function run_task that takes in some celery task.

@task(name='adder')
def add(x, y):
return x + y


def run_task(taskfn: "?", *a, **kw):
# do something here
taskfn.apply_async(args=a, kwargs=kw)
...


>>> run_task(add, 1, 2) # usage


My confusion lies here:

>>> from celery.task import Task
>>> from celery.local import PromiseProxy
>>> type(add).__name__
'PromiseProxy'
>>> isinstance(add, Task)
False
>>> isinstance(add, PromiseProxy)
True


It seems like PromiseProxy is some kind of "future" or "promisary" proxy that holds the task (or something like that, I haven't looked too far into it).

This wouldn't be an issue, I could easily do taskfn: PromiseProxy, but then taskfn.apply_async wouldn't make sense since there is no method apply_async on a PromiseProxy. So, my question is, should I do what type tells me to, or think in terms of duck-typing and go with Task?

Continue reading...