Last active
September 28, 2024 08:53
-
-
Save Kludex/776b685e4877903c6a59f97fc39d014b to your computer and use it in GitHub Desktop.
Properly annotate `apply_async` in Celery
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import TYPE_CHECKING | |
if TYPE_CHECKING: | |
from typing import Any, Callable, Generic, Literal, Protocol, Tuple, overload | |
from celery import Celery as _Celery | |
from celery import Task as _Task | |
from typing_extensions import TypeVarTuple, Unpack | |
Ts = TypeVarTuple("Ts") | |
class Task(Generic[Unpack[Ts]], _Task): | |
def apply_async( | |
self, | |
args: Tuple[Unpack[Ts]], | |
kwargs=None, | |
task_id=None, | |
producer=None, | |
link=None, | |
link_error=None, | |
shadow=None, | |
**options | |
): | |
... | |
class BindTaskFunc(Protocol[Unpack[Ts]]): | |
def __call__(self, __task: Task, *args: Unpack[Ts]) -> None: | |
... | |
class TaskFunc(Protocol[Unpack[Ts]]): | |
def __call__(self, *args: Unpack[Ts]) -> None: | |
... | |
class Celery(_Celery): | |
@overload | |
def task( | |
self, bind: Literal[True], *args: Any, **opts: Any | |
) -> Callable[[BindTaskFunc[Unpack[Ts]]], Task[Unpack[Ts]]]: | |
... | |
@overload | |
def task(self, *args: Any, **opts: Any) -> Callable[[TaskFunc[Unpack[Ts]]], Task[Unpack[Ts]]]: | |
... | |
def task(self, *args: Any, **opts: Any) -> ...: | |
... | |
else: | |
from celery import Celery | |
app = Celery() | |
@app.task() | |
def potato(id: str, another: int) -> None: | |
... | |
potato.apply_async(args=("1", 2)) | |
@app.task(bind=True) | |
def banana(self: Task, id: str, another: int) -> None: | |
... | |
banana.apply_async(args=("1", 2)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment