create new error class

Author: TheTechromancer

taskiq/abc/broker.py

@@ -29,7 +29,7 @@
 from taskiq.acks import AckableMessage
 from taskiq.decor import AsyncTaskiqDecoratedTask
 from taskiq.events import TaskiqEvents
-from taskiq.exceptions import TaskRejectedError
+from taskiq.exceptions import TaskBrokerMismatchError
 from taskiq.formatters.proxy_formatter import ProxyFormatter
 from taskiq.message import BrokerMessage
 from taskiq.result_backends.dummy import DummyResultBackend
@@ -518,14 +518,12 @@ def _register_task(
         By default we register tasks in local task registry.
         But this behaviour can be changed in subclasses.
 
-        This method may raise TaskRejectedError if task has already been
+        This method may raise TaskBrokerMismatchError if task has already been
         registered to a different broker.
 
         :param task_name: Name of a task.
         :param task: Decorated task.
         """
         if task.broker != self:
-            raise TaskRejectedError(
-                f"Task already has a different broker ({task.broker})",
-            )
+            raise TaskBrokerMismatchError(broker=task.broker)
         self.local_task_registry[task_name] = task

taskiq/exceptions.py

@@ -101,3 +101,9 @@ class ScheduledTaskCancelledError(TaskiqError):
     """Scheduled task was cancelled and not sent to the queue."""
 
     __template__ = "Cannot send scheduled task to the queue."
+
+
+class TaskBrokerMismatchError(TaskRejectedError):
+    """Task has a different broker than the one it was registered to."""
+
+    __template__ = "Task already has a different broker ({broker})"

taskiq/schedule_sources/label_based.py

@@ -32,7 +32,7 @@ async def get_schedules(self) -> List["ScheduledTask"]:
                 # if task broker doesn't match self, something is probably wrong
                 logger.warning(
                     f"Broker for {task_name} ({task.broker}) doesn't "
-                    f"match scheduler's broker ({self.broker})",
+                    f"match scheduler's broker ({self.broker})"
                 )
                 continue
             for schedule in task.labels.get("schedule", []):
@@ -70,8 +70,9 @@ def post_send(self, scheduled_task: ScheduledTask) -> None:
                 # if task broker doesn't match self, something is probably wrong
                 logger.warning(
                     f"Broker for {task_name} ({task.broker}) doesn't "
-                    f"match scheduler's broker ({self.broker})",
+                    f"match scheduler's broker ({self.broker})"
                 )
+                continue
             if scheduled_task.task_name != task_name:
                 continue