Add initial support for a celery backend

Author: matiasb

django_tasks/backends/celery/__init__.py

@@ -0,0 +1,3 @@
+from .backend import CeleryBackend
+
+__all__ = ["CeleryBackend"]

django_tasks/backends/celery/app.py

@@ -0,0 +1,21 @@
+import os
+
+from celery import Celery
+
+from django_tasks.task import DEFAULT_QUEUE_NAME
+
+
+# Set the default Django settings module for the 'celery' program.
+django_settings = os.environ.get('DJANGO_SETTINGS_MODULE')
+if django_settings is None:
+    raise ValueError('DJANGO_SETTINGS_MODULE environment variable is not set')
+
+app = Celery('django_tasks')
+
+# Using a string here means the worker doesn't have to serialize
+# the configuration object to child processes.
+# - namespace='CELERY' means all celery-related configuration keys
+#   should have a `CELERY_` prefix.
+app.config_from_object('django.conf:settings', namespace='CELERY')
+
+app.conf.task_default_queue = DEFAULT_QUEUE_NAME

django_tasks/backends/celery/backend.py

@@ -0,0 +1,43 @@
+from typing import TypeVar
+
+from typing_extensions import ParamSpec
+
+from celery import shared_task
+from celery.local import Proxy as CeleryTaskProxy
+from django_tasks.backends.base import BaseTaskBackend
+from django_tasks.task import Task, TaskResult
+
+
+T = TypeVar("T")
+P = ParamSpec("P")
+
+
+class CeleryTask(Task):
+
+    celery_task: CeleryTaskProxy
+    """Celery proxy to the task in the current celery app task registry."""
+
+    def __post_init__(self) -> None:
+        # TODO: allow passing extra celery specific parameters?
+        celery_task = shared_task()(self.func)
+        self.celery_task = celery_task
+        return super().__post_init__()
+
+
+class CeleryBackend(BaseTaskBackend):
+    task_class = CeleryTask
+    supports_defer = True
+
+    def enqueue(
+        self, task: Task[P, T], args: P.args, kwargs: P.kwargs
+    ) -> TaskResult[T]:
+        self.validate_task(task)
+
+        apply_async_kwargs = {
+            "eta": task.run_after,
+        }
+        if task.queue_name:
+            apply_async_kwargs["queue"] = task.queue_name
+        if task.priority:
+            apply_async_kwargs["priority"] = task.priority
+        task.celery_task.apply_async(args, kwargs=kwargs, **apply_async_kwargs)

pyproject.toml

@@ -54,6 +54,7 @@ dev = [
     "coverage",
     "django-stubs[compatible-mypy]",
     "dj-database-url",
+    "celery",
 ]
 mysql = [
     "mysqlclient"