celery 任务的正确类型注释

Question

我正在寻找正确的类型注释来归因于 celery 装饰器设置的 celery 任务@task。

假设我有一个函数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)