Added new exceptions to follow izulu flow

Signed-off-by: chandr-andr (Kiselev Aleksandr) <[email protected]>

Author: chandr-andr

taskiq/brokers/inmemory_broker.py

@@ -7,7 +7,7 @@
 from taskiq.abc.result_backend import AsyncResultBackend, TaskiqResult
 from taskiq.depends.progress_tracker import TaskProgress
 from taskiq.events import TaskiqEvents
-from taskiq.exceptions import TaskiqError
+from taskiq.exceptions import UnknownTaskError
 from taskiq.message import BrokerMessage
 from taskiq.receiver import Receiver
 from taskiq.utils import maybe_awaitable
@@ -156,7 +156,7 @@ async def kick(self, message: BrokerMessage) -> None:
         """
         target_task = self.find_task(message.task_name)
         if target_task is None:
-            raise TaskiqError(description="Unknown task.")
+            raise UnknownTaskError(task_name=message.task_name)
 
         receiver_cb = self.receiver.callback(message=message.message)
         if self.await_inplace:

taskiq/brokers/shared_broker.py

@@ -4,7 +4,7 @@
 
 from taskiq.abc.broker import AsyncBroker
 from taskiq.decor import AsyncTaskiqDecoratedTask
-from taskiq.exceptions import TaskiqError
+from taskiq.exceptions import SharedBrokerListenError, SharedBrokerSendTaskError
 from taskiq.kicker import AsyncKicker
 from taskiq.message import BrokerMessage
 
@@ -56,12 +56,7 @@ async def kick(self, message: BrokerMessage) -> None:
         :param message: message to send.
         :raises TaskiqError: if called.
         """
-        raise TaskiqError(
-            description=(
-                "You cannot use kiq directly on shared task "
-                "without setting the default_broker."
-            ),
-        )
+        raise SharedBrokerSendTaskError
 
     async def listen(self) -> AsyncGenerator[bytes, None]:  # type: ignore
         """
@@ -71,7 +66,7 @@ async def listen(self) -> AsyncGenerator[bytes, None]:  # type: ignore
 
         :raises TaskiqError: if called.
         """
-        raise TaskiqError(description="Shared broker cannot listen")
+        raise SharedBrokerListenError
 
     def _register_task(
         self,

taskiq/exceptions.py

@@ -23,12 +23,38 @@ class BrokerError(TaskiqError):
     __template__ = "Base exception for all broker errors"
 
 
+class ListenError(TaskiqError):
+    """Error if the broker is unable to listen to the queue."""
+
+
+class SharedBrokerListenError(ListenError):
+    """Error when someone tries to listen to the queue with shared broker."""
+
+    __template__ = "Shared broker cannot listen"
+
+
 class SendTaskError(BrokerError):
     """Error if the broker was unable to send the task to the queue."""
 
     __template__ = "Cannot send task to the queue"
 
 
+class SharedBrokerSendTaskError(SendTaskError):
+    """Error when someone tries to send task with shared broker."""
+
+    __template__ = (
+        "You cannot use kiq directly on shared task "
+        "without setting the default_broker."
+    )
+
+
+class UnknownTaskError(SendTaskError):
+    """Error if task is unknown."""
+
+    __template__ = "Cannot send unknown task to the queue, task name - {task_name}"
+    task_name: str
+
+
 class ResultBackendError(TaskiqError):
     """Base class for all ResultBackend errors."""