fix: clean up lint errors and typos in observability code

- Fix typos in observer docstring and metric descriptions
- Add missing docstrings to observer protocol and implementation methods
- Remove unused Gauge import from PrometheusMiddleware.__init__
- Remove unused ReceiverObserver import from run.py
- Fix import ordering (ruff I001)
- Add noqa for expected complexity in runner method
- Run black formatting

Author: Mohammed0tarek

taskiq/cli/worker/run.py

@@ -13,7 +13,7 @@
 from taskiq.cli.utils import import_object, import_tasks
 from taskiq.cli.worker.args import WorkerArgs
 from taskiq.cli.worker.process_manager import ProcessManager
-from taskiq.receiver import Receiver, ReceiverObserver
+from taskiq.receiver import Receiver
 
 try:
     import uvloop

taskiq/middlewares/prometheus_middleware.py

@@ -4,10 +4,11 @@
 from pathlib import Path
 from tempfile import gettempdir
 from typing import Any
+
 from taskiq.abc.middleware import TaskiqMiddleware
 from taskiq.message import TaskiqMessage
-from taskiq.result import TaskiqResult
 from taskiq.receiver.observer import ReceiverObserver
+from taskiq.result import TaskiqResult
 
 logger = getLogger("taskiq.prometheus")
 
@@ -21,7 +22,7 @@ class PrometheusMiddleware(TaskiqMiddleware):
 
     :param server_port: The port to listen on.
     :param server_addr: The address to listen on.
-    :paam metrics_path: The path to store metrics for multiproc env.
+    :param metrics_path: The path to store metrics for multiproc env.
     """
 
     def __init__(
@@ -44,7 +45,7 @@ def __init__(
         logger.debug("Initializing metrics")
 
         try:
-            from prometheus_client import Counter, Histogram, Gauge  # noqa: PLC0415
+            from prometheus_client import Counter, Histogram  # noqa: PLC0415
         except ImportError as exc:
             raise ImportError(
                 "Cannot initialize metrics. Please install 'taskiq[metrics]'.",
@@ -199,9 +200,14 @@ def post_execute(
             self.success_tasks.labels(message.task_name).inc()
         self.execution_time.labels(message.task_name).observe(result.execution_time)
 
-    def set_broker(self, broker: "AsyncBroker") -> None:  # noqa: F821 pyright: ignore[reportUnknownVariableType]
+    def set_broker(self, broker: "AsyncBroker") -> None:  # noqa: F821
+        """
+        Set broker and attach receiver observer.
+
+        :param broker: broker to set.
+        """
         super().set_broker(broker)
-        broker._receiver_observer = PrometheusReceiverObserver()
+        broker._receiver_observer = PrometheusReceiverObserver()  # noqa: SLF001
 
     def post_save(
         self,
@@ -250,20 +256,25 @@ def __init__(self) -> None:
         )
         self.deserialize_error = Counter(
             "deserialize_error_count",
-            "Number of times broker faced a desrialization error",
+            "Number of times broker faced a deserialization error",
         )
 
     def on_prefetch_queue_size(self, size: int) -> None:
+        """Record current prefetch queue depth."""
         self.prefetch_queue_size.set(size)
 
     def on_semaphore_status(self, available: int) -> None:
+        """Record available semaphore slots."""
         self.semaphore_available.set(available)
 
     def on_active_tasks_count(self, count: int) -> None:
+        """Record number of currently executing tasks."""
         self.active_tasks_count.set(count)
 
     def on_task_not_found(self, task_name: str) -> None:
+        """Increment counter for unregistered task lookups."""
         self.task_not_found_total.labels(task_name).inc()
 
     def on_deserialize_error(self, raw: bytes, error: Exception) -> None:
+        """Increment counter for message deserialization failures."""
         self.deserialize_error.inc()

taskiq/receiver/observer.py

@@ -4,18 +4,32 @@
 @runtime_checkable
 class ReceiverObserver(Protocol):
     """
-    Observer for reciever stats.
+    Observer for receiver stats.
 
-    This classs is used to observe/collect metrics for the receiver.
+    This class is used to observe/collect metrics for the receiver.
     This includes semaphore usage, tasks in queue, etc.
 
     metrics tracked:
     - Number of tasks in queue
-    - Number of taks in execution (Semaphore uusage)
+    - Number of tasks in execution (semaphore usage)
     """
 
-    def on_prefetch_queue_size(self, size: int) -> None: ...
-    def on_semaphore_status(self, available: int) -> None: ...
-    def on_active_tasks_count(self, count: int) -> None: ...
-    def on_task_not_found(self, task_name: str) -> None: ...
-    def on_deserialize_error(self, raw: bytes, error: Exception) -> None: ...
+    def on_prefetch_queue_size(self, size: int) -> None:
+        """Called when the prefetch queue size changes."""
+        ...
+
+    def on_semaphore_status(self, available: int) -> None:
+        """Called when semaphore availability changes."""
+        ...
+
+    def on_active_tasks_count(self, count: int) -> None:
+        """Called when the number of active tasks changes."""
+        ...
+
+    def on_task_not_found(self, task_name: str) -> None:
+        """Called when a received task is not registered."""
+        ...
+
+    def on_deserialize_error(self, raw: bytes, error: Exception) -> None:
+        """Called when a message fails to deserialize."""
+        ...

taskiq/receiver/receiver.py

@@ -412,7 +412,7 @@ async def prefetcher(
         await queue.put(QUEUE_DONE)
         self.sem_prefetch.release()
 
-    async def runner(
+    async def runner(  # noqa: C901
         self,
         queue: "asyncio.Queue[bytes | AckableMessage]",
     ) -> None:
@@ -443,26 +443,20 @@ def task_cb(task: "asyncio.Task[Any]") -> None:
                 self.sem.release()
 
                 if self.observer is not None:
-                    self.observer.on_semaphore_status(
-                        self.sem._value  # noqa
-                    )
+                    self.observer.on_semaphore_status(self.sem._value)  # noqa
 
         while True:
             try:
                 # Waits for semaphore to be released.
                 if self.sem is not None:
                     await self.sem.acquire()
                     if self.observer is not None:
-                        self.observer.on_semaphore_status(
-                            self.sem._value  # noqa
-                        )
+                        self.observer.on_semaphore_status(self.sem._value)  # noqa
 
                 self.sem_prefetch.release()
                 message = await queue.get()
                 if self.observer is not None:
-                    self.observer.on_prefetch_queue_size(
-                        queue.qsize()  # noqa
-                    )
+                    self.observer.on_prefetch_queue_size(queue.qsize())
                 if message is QUEUE_DONE:
                     # asyncio.wait will throw an error if there is nothing to wait for
                     if tasks: