Define a default queue name

Author: RealOrangeOne

django_tasks/__init__.py

@@ -6,11 +6,12 @@
 
 from .backends.base import BaseTaskBackend
 from .exceptions import InvalidTaskBackendError
-from .task import ResultStatus, Task, task
+from .task import DEFAULT_QUEUE_NAME, ResultStatus, Task, task
 
 __all__ = [
     "tasks",
     "DEFAULT_TASK_BACKEND_ALIAS",
+    "DEFAULT_QUEUE_NAME",
     "task",
     "ResultStatus",
     "Task",

django_tasks/mail.py

@@ -23,6 +23,7 @@ class SMTPEmailBackend(BaseEmailBackend):
     """
     An email backend to send emails as background tasks
     """
+
     @staticmethod
     def serialize_message(message: EmailMessage) -> dict:
         data = {

django_tasks/task.py

@@ -12,6 +12,8 @@
 if TYPE_CHECKING:
     from .backends.base import BaseTaskBackend
 
+DEFAULT_QUEUE_NAME = "default"
+
 
 class ResultStatus(TextChoices):
     NEW = "NEW"
@@ -32,12 +34,12 @@ class Task(Generic[P, T]):
     func: Callable[P, T]
     """The task function"""
 
-    queue_name: Optional[str]
-    """The name of the queue the task will run on """
-
     backend: str
     """The name of the backend the task will run on"""
 
+    queue_name: str = DEFAULT_QUEUE_NAME
+    """The name of the queue the task will run on """
+
     run_after: Optional[datetime] = None
     """The earliest this task will run"""
 
@@ -126,7 +128,7 @@ def get_backend(self) -> "BaseTaskBackend":
 
 def task(
     priority: Optional[int] = None,
-    queue_name: Optional[str] = None,
+    queue_name: str = DEFAULT_QUEUE_NAME,
     backend: str = "default",
 ) -> Callable[[Callable[P, T]], Task[P, T]]:
     """

tests/tests/test_tasks.py

@@ -4,7 +4,13 @@
 from django.test import SimpleTestCase, override_settings
 from django.utils import timezone
 
-from django_tasks import ResultStatus, default_task_backend, task, tasks
+from django_tasks import (
+    DEFAULT_QUEUE_NAME,
+    ResultStatus,
+    default_task_backend,
+    task,
+    tasks,
+)
 from django_tasks.backends.dummy import DummyBackend
 from django_tasks.backends.immediate import ImmediateBackend
 from django_tasks.exceptions import (
@@ -56,11 +62,11 @@ def test_using_priority(self) -> None:
         self.assertIsNone(test_tasks.noop_task.priority)
 
     def test_using_queue_name(self) -> None:
-        self.assertIsNone(test_tasks.noop_task.queue_name)
+        self.assertEqual(test_tasks.noop_task.queue_name, DEFAULT_QUEUE_NAME)
         self.assertEqual(
             test_tasks.noop_task.using(queue_name="queue_1").queue_name, "queue_1"
         )
-        self.assertIsNone(test_tasks.noop_task.queue_name)
+        self.assertEqual(test_tasks.noop_task.queue_name, DEFAULT_QUEUE_NAME)
 
     def test_using_run_after(self) -> None:
         now = timezone.now()