app without a task type specializes to Task[P, R]

celery-stubs/app/base.pyi

@@ -201,6 +201,21 @@ class Celery(Generic[_T_Global]):
     def close(self) -> None: ...
     def start(self, argv: list[str] | None = None) -> NoReturn: ...
     def worker_main(self, argv: list[str] | None = None) -> NoReturn: ...
+
+    # There are 4 independent parts of the task decorator that we want to
+    # cover with overloads, leading to some repetition in function signatures.
+    # The distinctions we want to cover are:
+    # - Does the Celery application specify a base task type `_T_Global`
+    # - is the app.task called as a decorator
+    #   `def task(self, *, ...) -> Callable[[Callable], Task]`
+    # or a constructor
+    #   `def task(self, function, *, ...) -> Task`
+    # - Does the task receive a self parameter through specifying `bind=True`
+    # - Does the task decorator specify a base task type _T through `base=_T`
+    @overload
+    def task(
+        self: Celery[CeleryTask[Any, Any]], fun: Callable[_P, _R]
+    ) -> CeleryTask[_P, _R]: ...
     @overload
     def task(self, fun: Callable[_P, _R]) -> _T_Global: ...
     @overload
@@ -244,6 +259,48 @@ class Celery(Generic[_T_Global]):
         **options: Any,
     ) -> Callable[[Callable[Concatenate[_T, _P], _R]], _T]: ...
     @overload
+    def task(
+        self: Celery[CeleryTask[Any, Any]],
+        *,
+        name: str = ...,
+        serializer: str = ...,
+        bind: Literal[True],
+        autoretry_for: Sequence[type[BaseException]] = ...,
+        dont_autoretry_for: Sequence[type[BaseException]] = ...,
+        max_retries: int | None = ...,
+        default_retry_delay: int = ...,
+        acks_late: bool = ...,
+        ignore_result: bool = ...,
+        soft_time_limit: float | None = ...,
+        time_limit: float | None = ...,
+        base: None = ...,
+        retry_kwargs: dict[str, Any] = ...,
+        retry_backoff: bool | int = ...,
+        retry_backoff_max: int = ...,
+        retry_jitter: bool = ...,
+        typing: bool = ...,
+        rate_limit: str | None = ...,
+        trail: bool = ...,
+        send_events: bool = ...,
+        store_errors_even_if_ignored: bool = ...,
+        autoregister: bool = ...,
+        track_started: bool = ...,
+        acks_on_failure_or_timeout: bool = ...,
+        reject_on_worker_lost: bool = ...,
+        throws: tuple[type[Exception], ...] = ...,
+        expires: float | datetime.datetime | None = ...,
+        priority: int | None = ...,
+        resultrepr_maxsize: int = ...,
+        request_stack: _LocalStack[Context] = ...,
+        abstract: bool = ...,
+        queue: str = ...,
+        after_return: Callable[..., Any] = ...,
+        on_retry: Callable[..., Any] = ...,
+        **options: Any,
+    ) -> Callable[
+        [Callable[Concatenate[CeleryTask[Any, Any], _P], _R]], CeleryTask[_P, _R]
+    ]: ...
+    @overload
     def task(
         self,
         *,
@@ -284,6 +341,46 @@ class Celery(Generic[_T_Global]):
         **options: Any,
     ) -> Callable[[Callable[Concatenate[_T_Global, _P], _R]], _T_Global]: ...
     @overload
+    def task(
+        self: Celery[CeleryTask[Any, Any]],
+        *,
+        name: str = ...,
+        serializer: str = ...,
+        bind: Literal[False] = False,
+        autoretry_for: Sequence[type[BaseException]] = ...,
+        dont_autoretry_for: Sequence[type[BaseException]] = ...,
+        max_retries: int | None = ...,
+        default_retry_delay: int = ...,
+        acks_late: bool = ...,
+        ignore_result: bool = ...,
+        soft_time_limit: float | None = ...,
+        time_limit: float | None = ...,
+        base: None = ...,
+        retry_kwargs: dict[str, Any] = ...,
+        retry_backoff: bool | int = ...,
+        retry_backoff_max: int = ...,
+        retry_jitter: bool = ...,
+        typing: bool = ...,
+        rate_limit: str | None = ...,
+        trail: bool = ...,
+        send_events: bool = ...,
+        store_errors_even_if_ignored: bool = ...,
+        autoregister: bool = ...,
+        track_started: bool = ...,
+        acks_on_failure_or_timeout: bool = ...,
+        reject_on_worker_lost: bool = ...,
+        throws: tuple[type[Exception], ...] = ...,
+        expires: float | datetime.datetime | None = ...,
+        priority: int | None = ...,
+        resultrepr_maxsize: int = ...,
+        request_stack: _LocalStack[Context] = ...,
+        abstract: bool = ...,
+        queue: str = ...,
+        after_return: Callable[..., Any] = ...,
+        on_retry: Callable[..., Any] = ...,
+        **options: Any,
+    ) -> Callable[[Callable[_P, _R]], CeleryTask[_P, _R]]: ...
+    @overload
     def task(
         self,
         *,

tests/test_celery.py

@@ -2,7 +2,7 @@
 
 import errno
 import sys
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Any, ParamSpec, Protocol, TypeVar
 
 import celery
 from celery import Celery, shared_task, signature
@@ -15,11 +15,13 @@
 from typing_extensions import assert_type, override
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Callable, Iterator
 
     from celery.contrib.abortable import AbortableTask
     from celery.contrib.django.task import DjangoTask
 
+P = ParamSpec("P")
+R = TypeVar("R")
 app = celery.Celery()
 
 logger = get_task_logger(__name__)
@@ -40,6 +42,43 @@ def sub(x: int, y: int) -> int:
     return x - y
 
 
+def test_task_constructor_signature(func: Callable[P, R]) -> None:
+    """
+    The constructor form of the task creation decorator passes the function
+    signature through to the newly created task.
+    """
+    task_from_constructor = app.task(func)
+    assert_type(task_from_constructor, Task[P, R])
+
+
+def test_task_decorator_signature(func: Callable[P, R]) -> None:
+    """
+    The task creation decorator passes the function signature through to the
+    newly created task.
+    """
+
+    @app.task(max_retries=1)
+    def signature_passthrough(*args: P.args, **kwargs: P.kwargs) -> R:
+        return func(*args, **kwargs)
+
+    assert_type(signature_passthrough, Task[P, R])
+
+
+def test_bound_task_decorator_signature(func: Callable[P, R]) -> None:
+    """
+    The bound task creation decorator creates a new task with a function
+    signature that does not include the first `self` parameter.
+    """
+
+    @app.task(bind=True)
+    def signature_passthrough(
+        self: Task[Any, Any], *args: P.args, **kwargs: P.kwargs
+    ) -> R:
+        return func(*args, **kwargs)
+
+    assert_type(signature_passthrough, Task[P, R])
+
+
 class Table(Protocol):
     def all(self) -> Iterator[dict[str, object]]: ...
 
@@ -81,7 +120,7 @@ def process_rows_3(self: DatabaseTask, param_1: int) -> None:
 
 # Here, a typeignore is needed so that when the overload stops working correctly,
 # pyright and mypy will report that the typeignore is unnecessary.
-@shared_task(base=DatabaseTask, bind=True)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
+@shared_task(base=DatabaseTask, bind=True)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]  # ty: ignore[invalid-argument-type]
 def process_rows_4(self: int, param_1: int) -> None:
     assert_type(process_rows_4, DatabaseTask)
 
@@ -109,7 +148,7 @@ def process_rows_6(self: DatabaseTask, param_1: int) -> None:
 
 # Here, a typeignore is needed so that when the overload stops working correctly,
 # pyright and mypy will report that the typeignore is unnecessary.
-@database_app.task(name="main.process_rows_7", bind=True)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
+@database_app.task(name="main.process_rows_7", bind=True)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]  # ty: ignore[invalid-argument-type]
 def process_rows_7(self: int, param_1: int) -> None:
     assert_type(process_rows_7, DatabaseTask)
 
@@ -119,7 +158,7 @@ def process_rows_7(self: int, param_1: int) -> None:
 
 # Here, a typeignore is needed so that when the overload stops working correctly,
 # pyright and mypy will report that the typeignore is unnecessary.
-@database_app.task(name="main.process_rows_8", bind=True, base=Task[..., None])  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
+@database_app.task(name="main.process_rows_8", bind=True, base=Task[..., None])  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]  # ty: ignore[invalid-argument-type]
 def binded_task_8_fail(self: int, param_1: int) -> None:
     pass