celery 任务的正确类型注释

Question

我正在寻找正确的类型注释来归因于已由 celery

@task

装饰器设置的 celery 任务。

假设我有一个函数

run_task

可以执行一些 celery 任务。

@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

我的困惑就在这里：

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

看起来

PromiseProxy

是某种“未来”或“承诺”的代理，它持有任务（或类似的东西，我还没有深入研究过它）。

这不是问题，我可以轻松做到

taskfn: PromiseProxy

，但是

taskfn.apply_async

就没有意义，因为

apply_async

上没有方法

PromiseProxy

。所以，我的问题是，我应该按照

type

告诉我的去做，还是按照鸭子打字的方式思考并选择

Task

？

Answer 1

我这样做只是为了IDE检测延迟方法

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)