Correct type annotation for a celery task

Question

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?

Answer 1

I did it like this just for IDE detecting delay method

from celery import shared_task
from celery.app.task import Task


def my_task(compound_ids: list[int] | None = None):
    print(1 + 1)


my_task: Task = shared_task(my_task)