Django 6.0: Add Django Tasks framework

django-stubs/tasks/__init__.pyi

@@ -0,0 +1,28 @@
+from django.utils.connection import BaseConnectionHandler
+
+from .backends.base import BaseTaskBackend
+from .base import DEFAULT_TASK_BACKEND_ALIAS as DEFAULT_TASK_BACKEND_ALIAS
+from .base import DEFAULT_TASK_QUEUE_NAME as DEFAULT_TASK_QUEUE_NAME
+from .base import Task as Task
+from .base import TaskContext as TaskContext
+from .base import TaskResult as TaskResult
+from .base import TaskResultStatus as TaskResultStatus
+from .base import task as task
+
+__all__ = [
+    "DEFAULT_TASK_BACKEND_ALIAS",
+    "DEFAULT_TASK_QUEUE_NAME",
+    "Task",
+    "TaskContext",
+    "TaskResult",
+    "TaskResultStatus",
+    "default_task_backend",
+    "task",
+    "task_backends",
+]
+
+class TaskBackendHandler(BaseConnectionHandler[BaseTaskBackend]): ...
+
+task_backends: TaskBackendHandler
+# Actually ConnectionProxy, but quacks exactly like BaseTaskBackend, it's not worth distinguishing the two.
+default_task_backend: BaseTaskBackend

django-stubs/tasks/backends/base.pyi

@@ -0,0 +1,21 @@
+from abc import ABCMeta
+from typing import Any
+
+from django.tasks.base import Task, TaskResult
+
+class BaseTaskBackend(metaclass=ABCMeta):
+    task_class: type[Task]
+    supports_defer: bool
+    supports_async_task: bool
+    supports_get_result: bool
+    supports_priority: bool
+    alias: str
+    queues: list[str]
+    options: dict[str, Any]
+    def __init__(self, alias: str, params: dict[str, Any]) -> None: ...
+    def validate_task(self, task: Task) -> None: ...
+    def enqueue(self, task: Task, args: list[Any], kwargs: dict[str, Any]) -> TaskResult: ...
+    async def aenqueue(self, task: Task, args: list[Any], kwargs: dict[str, Any]) -> TaskResult: ...
+    def get_result(self, result_id: str) -> TaskResult: ...
+    async def aget_result(self, result_id: str) -> TaskResult: ...
+    def check(self, **kwargs: Any) -> list[Any]: ...

django-stubs/tasks/backends/dummy.pyi

@@ -0,0 +1,13 @@
+from typing import Any
+
+from django.tasks.base import Task, TaskResult
+
+from .base import BaseTaskBackend
+
+class DummyBackend(BaseTaskBackend):
+    results: list[TaskResult]
+    def __init__(self, alias: str, params: dict[str, Any]) -> None: ...
+    def enqueue(self, task: Task, args: list[Any], kwargs: dict[str, Any]) -> TaskResult: ...
+    def get_result(self, result_id: str) -> TaskResult: ...
+    async def aget_result(self, result_id: str) -> TaskResult: ...
+    def clear(self) -> None: ...

django-stubs/tasks/backends/immediate.pyi

@@ -0,0 +1,13 @@
+from logging import Logger
+from typing import Any
+
+from django.tasks.base import Task, TaskResult
+
+from .base import BaseTaskBackend as BaseTaskBackend
+
+logger: Logger
+
+class ImmediateBackend(BaseTaskBackend):
+    worker_id: str
+    def __init__(self, alias: str, params: dict[str, Any]) -> None: ...
+    def enqueue(self, task: Task, args: list[Any], kwargs: dict[str, Any]) -> TaskResult: ...

django-stubs/tasks/base.pyi

@@ -0,0 +1,111 @@
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Generic, TypeVar, overload
+
+from django.db.models.enums import TextChoices as TextChoices
+from django.utils.module_loading import import_string as import_string
+from django.utils.translation import pgettext_lazy as pgettext_lazy
+from typing_extensions import ParamSpec
+
+from .backends.base import BaseTaskBackend
+from .exceptions import TaskResultMismatch as TaskResultMismatch
+
+DEFAULT_TASK_BACKEND_ALIAS: str
+DEFAULT_TASK_PRIORITY: int
+DEFAULT_TASK_QUEUE_NAME: str
+TASK_MAX_PRIORITY: int
+TASK_MIN_PRIORITY: int
+TASK_REFRESH_ATTRS: set[str]
+
+class TaskResultStatus(TextChoices):
+    READY = ...
+    RUNNING = ...
+    FAILED = ...
+    SUCCESSFUL = ...
+
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+
+@dataclass(kw_only=True)
+class Task(Generic[_P, _R]):
+    priority: int
+    func: Callable[_P, _R]
+    backend: str
+    queue_name: str
+    run_after: datetime | None
+    takes_context: bool = ...
+    def __post_init__(self) -> None: ...
+    @property
+    def name(self) -> str: ...
+    def using(
+        self,
+        *,
+        priority: int | None = None,
+        queue_name: str | None = None,
+        run_after: datetime | None = None,
+        backend: str | None = None,
+    ) -> Task[_P, _R]: ...
+    def enqueue(self, *args: _P.args, **kwargs: _P.kwargs) -> TaskResult[_P, _R]: ...
+    async def aenqueue(self, *args: _P.args, **kwargs: _P.kwargs) -> TaskResult[_P, _R]: ...
+    def get_result(self, result_id: str) -> TaskResult[_P, _R]: ...
+    async def aget_result(self, result_id: str) -> TaskResult[_P, _R]: ...
+    def call(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: ...
+    async def acall(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: ...
+    def get_backend(self) -> BaseTaskBackend: ...
+    @property
+    def module_path(self) -> str: ...
+
+@overload
+def task(
+    function: Callable[_P, _R],
+    *,
+    priority: int | None = None,
+    queue_name: str | None = None,
+    backend: str | None = None,
+    takes_context: bool = ...,
+) -> Task[_P, _R]: ...
+@overload
+def task(
+    *,
+    priority: int | None = None,
+    queue_name: str | None = None,
+    backend: str | None = None,
+    takes_context: bool = ...,
+) -> Callable[[Callable[_P, _R]], Task[_P, _R]]: ...
+@dataclass(kw_only=True)
+class TaskError:
+    exception_class_path: str
+    traceback: str
+    @property
+    def exception_class(self) -> type[BaseException]: ...
+
+@dataclass(kw_only=True)
+class TaskResult(Generic[_P, _R]):
+    task: Task[_P, _R]
+    id: str
+    status: TaskResultStatus
+    enqueued_at: datetime | None
+    started_at: datetime | None
+    finished_at: datetime | None
+    last_attempted_at: datetime | None
+    args: list[Any]
+    kwargs: dict[str, Any]
+    backend: str
+    errors: list[TaskError]
+    worker_ids: list[str]
+    def __post_init__(self) -> None: ...
+    @property
+    def return_value(self) -> _R: ...
+    @property
+    def is_finished(self) -> bool: ...
+    @property
+    def attempts(self) -> int: ...
+    def refresh(self) -> None: ...
+    async def arefresh(self) -> None: ...
+
+@dataclass(kw_only=True)
+class TaskContext(Generic[_P, _R]):
+    task_result: TaskResult[_P, _R]
+    @property
+    def attempt(self) -> int: ...

django-stubs/tasks/exceptions.pyi

@@ -0,0 +1,7 @@
+from django.core.exceptions import ImproperlyConfigured as ImproperlyConfigured
+
+class TaskException(Exception): ...
+class InvalidTask(TaskException): ...
+class InvalidTaskBackend(ImproperlyConfigured): ...
+class TaskResultDoesNotExist(TaskException): ...
+class TaskResultMismatch(TaskException): ...

django-stubs/tasks/signals.pyi

@@ -0,0 +1,19 @@
+from logging import Logger
+from typing import Any
+
+from django.dispatch import Signal as Signal
+from django.tasks.backends.base import BaseTaskBackend
+from django.tasks.base import TaskResult
+
+from .base import TaskResultStatus as TaskResultStatus
+
+logger: Logger
+
+task_enqueued: Signal
+task_finished: Signal
+task_started: Signal
+
+def clear_tasks_handlers(*, setting: str, **kwargs: Any) -> None: ...
+def log_task_enqueued(sender: type[BaseTaskBackend], task_result: TaskResult, **kwargs: Any) -> None: ...
+def log_task_started(sender: type[BaseTaskBackend], task_result: TaskResult, **kwargs: Any) -> None: ...
+def log_task_finished(sender: type[BaseTaskBackend], task_result: TaskResult, **kwargs: Any) -> None: ...

ext/django_stubs_ext/patch.py

@@ -100,6 +100,17 @@ def __repr__(self) -> str:
     MPGeneric(Prefetch),
 ]
 
+if VERSION >= (6, 0):
+    from django.tasks import Task, TaskContext, TaskResult
+
+    _need_generic.extend(
+        [
+            MPGeneric(Task),
+            MPGeneric(TaskContext),
+            MPGeneric(TaskResult),
+        ]
+    )
+
 
 def monkeypatch(extra_classes: Iterable[type] | None = None, include_builtins: bool = True) -> None:
     """Monkey patch django as necessary to work properly with mypy."""

scripts/stubtest/allowlist_todo_django60.txt

@@ -202,15 +202,12 @@ django.forms.ClearableFileInput.use_fieldset
 django.forms.models.BaseModelForm.validate_constraints
 django.forms.renderers.Jinja2DivFormRenderer
 django.forms.widgets.ClearableFileInput.use_fieldset
-django.tasks
-django.tasks.backends
-django.tasks.backends.base
-django.tasks.backends.dummy
-django.tasks.backends.immediate
-django.tasks.base
 django.tasks.checks
-django.tasks.exceptions
-django.tasks.signals
+django.tasks.default_task_backend
+django.tasks.task
+django.tasks.backends.base.BaseTaskBackend.enqueue
+django.tasks.backends.immediate.BaseTaskBackend.enqueue
+django.tasks.base.task
 django.template.base.PartialTemplate
 django.template.defaulttags.PartialDefNode
 django.template.defaulttags.PartialNode

tests/test_generic_consistency.py

@@ -61,6 +61,9 @@ def test_find_classes_inheriting_from_generic() -> None:
         # For each Generic in the stubs, import the associated module and capture every class in the MRO
         if generic_visitor.generic_classes:
             module_name = _get_module_from_pyi(file_path)
+            # Skip tasks modules for Django < 6.0 because it's not available
+            if module_name.startswith("django.tasks") and django.VERSION < (6, 0):
+                continue
             django_module = importlib.import_module(module_name)
             all_generic_classes.update(
                 {

tests/typecheck/tasks/test_tasks.yaml

@@ -0,0 +1,23 @@
+-   case: task
+    main: |
+        from typing_extensions import reveal_type
+        from django.tasks import task
+
+        @task
+        def test_task(x: int) -> int:
+            return x + 1
+
+        reveal_type(test_task) # N: Revealed type is "django.tasks.base.Task[[x: builtins.int], builtins.int]"
+        reveal_type(test_task.call) # N: Revealed type is "def (x: builtins.int) -> builtins.int"
+        reveal_type(test_task.enqueue) # N: Revealed type is "def (x: builtins.int) -> django.tasks.base.TaskResult[[x: builtins.int], builtins.int]"
+
+-   case: task_with_priority
+    main: |
+        from typing_extensions import reveal_type
+        from django.tasks import task
+
+        @task(priority=5)
+        def test_task(x: int) -> int:
+            return x + 1
+
+        reveal_type(test_task) # N: Revealed type is "django.tasks.base.Task[[x: builtins.int], builtins.int]"