Skip to content

Instantly share code, notes, and snippets.

@Kludex
Last active September 28, 2024 08:53
Show Gist options
  • Save Kludex/776b685e4877903c6a59f97fc39d014b to your computer and use it in GitHub Desktop.
Save Kludex/776b685e4877903c6a59f97fc39d014b to your computer and use it in GitHub Desktop.
Properly annotate `apply_async` in Celery
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