Add type annotations across the workflows package

Annotate Recipe, RecipeWrapper, CommonTransport, common_service,
transport middleware (incl. OTEL), pika/stomp/offline transports,
frontend, contrib, services, and util/zocalo modules. Also includes
related minor refactors required by typing: elevate kwargs to explicit
parameters where appropriate; make CommonService.transport raise if
unset; route the cls.recipe override correctly in Recipe.__init__;
forward log_extender in recipe.wrap_subscribe; type-annotate
validate_recipe and the offline/middleware classes; tighten test
signature expectations to match the new exact send() signature.

Author: ndevenish

.pre-commit-config.yaml

@@ -10,6 +10,7 @@ repos:
   - id: check-merge-conflict
   - id: check-ast
     fail_fast: True
+    language_version: python3.12
   - id: check-json
   - id: check-added-large-files
     args: ['--maxkb=200']
@@ -35,4 +36,5 @@ repos:
   hooks:
   - id: mypy
     files: 'src/.*\.py$'
-    additional_dependencies: ['types-setuptools==57.0.2']
+    additional_dependencies: ['types-setuptools>=57.0.2']
+    language_version: python3.12

pyproject.toml

@@ -110,3 +110,6 @@ required-imports = ["from __future__ import annotations"]
 
 [tool.mypy]
 mypy_path = "src/"
+
+[tool.pyright]
+exclude = ["tests", ".venv"]

src/workflows/contrib/start_service.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import optparse
 import sys
 from collections.abc import Callable
 from optparse import SUPPRESS_HELP, OptionParser
+from typing import Any
 
 import workflows
 import workflows.frontend
@@ -17,20 +19,22 @@ class ServiceStarter:
     used in a number of scenarios."""
 
     @staticmethod
-    def on_parser_preparation(parser):
+    def on_parser_preparation(parser: OptionParser) -> OptionParser | None:
         """Plugin hook to manipulate the OptionParser object before command line
         parsing. If a value is returned here it will replace the OptionParser
         object."""
 
     @staticmethod
-    def on_parsing(options, args):
+    def on_parsing(
+        options: optparse.Values, args: list[str]
+    ) -> tuple[optparse.Values, list[str]] | None:
         """Plugin hook to manipulate the command line parsing results.
         A tuple of values can be returned, which will replace (options, args).
         """
 
     @staticmethod
     def on_transport_factory_preparation(
-        transport_factory,
+        transport_factory: Callable[[], CommonTransport],
     ) -> Callable[[], CommonTransport] | None:
         """Plugin hook to intercept/manipulate newly created Transport factories
         before first invocation."""
@@ -41,24 +45,26 @@ def on_transport_preparation(transport: CommonTransport) -> CommonTransport | No
         before connecting."""
 
     @staticmethod
-    def before_frontend_construction(kwargs):
+    def before_frontend_construction(kwargs: dict[str, Any]) -> dict[str, Any] | None:
         """Plugin hook to manipulate the Frontend object constructor arguments. If
         a value is returned here it will replace the keyword arguments
         dictionary passed to the constructor."""
 
     @staticmethod
-    def on_frontend_preparation(frontend):
+    def on_frontend_preparation(
+        frontend: workflows.frontend.Frontend,
+    ) -> workflows.frontend.Frontend | None:
         """Plugin hook to manipulate the Frontend object before starting it. If a
         value is returned here it will replace the Frontend object."""
 
     def run(
         self,
-        cmdline_args=None,
-        program_name="start_service",
-        version=None,
+        cmdline_args: list[str] | None = None,
+        program_name: str = "start_service",
+        version: str | None = None,
         add_metrics_option: bool = False,
-        **kwargs,
-    ):
+        **kwargs: Any,
+    ) -> None:
         """Example command line interface to start services.
 
         Args:
@@ -165,10 +171,12 @@ def on_transport_preparation_hook() -> CommonTransport:
         if options.service not in known_services:
             # First check whether the provided service name is a case-insensitive match.
             service_lower = options.service.lower()
-            match = {s.lower(): s for s in known_services}.get(service_lower, None)
-            match = (
-                [match]
-                if match
+            exact_match = {s.lower(): s for s in known_services}.get(
+                service_lower, None
+            )
+            match: list[str] = (
+                [exact_match]
+                if exact_match
                 # Next, check whether the provided service name is a partial
                 # case-sensitive match.
                 else [s for s in known_services if s.startswith(options.service)]

src/workflows/contrib/status_monitor.py

@@ -3,12 +3,12 @@
 import curses
 import threading
 import time
-from typing import Any
+from collections.abc import Mapping
+from typing import Any, Callable
 
 import workflows.transport
 from workflows.services.common_service import CommonService
-
-basestring = (str, bytes)
+from workflows.transport.common_transport import CommonTransport
 
 
 class Monitor:  # pragma: no cover
@@ -19,29 +19,29 @@ class Monitor:  # pragma: no cover
     shutdown = False
     """Set to true to end the main loop and shut down the service monitor."""
 
-    cards: dict[Any, Any] = {}
+    cards: list
     """Register card shown for seen services"""
 
     border_chars = ()
     """Characters used for frame borders."""
     border_chars_text = ("|", "|", "=", "=", "/", "\\", "\\", "/")
     """Example alternative set of frame border characters."""
 
-    def __init__(self, transport=None):
+    def __init__(self, transport: Callable[[], CommonTransport] | str | None = None):
         """Set up monitor and connect to the network transport layer"""
-        if transport is None or isinstance(transport, basestring):
-            self._transport = workflows.transport.lookup(transport)()
-        else:
+        if callable(transport):
             self._transport = transport()
+        else:
+            self._transport = workflows.transport.lookup(transport)()
         assert self._transport.connect(), "Could not connect to transport layer"
         self._lock = threading.RLock()
-        self._node_status = {}
-        self.message_box = None
+        self._node_status: dict = {}
+        self.message_box: curses.window | None = None
         self._transport.subscribe_broadcast(
             "transient.status", self.update_status, retroactive=True
         )
 
-    def update_status(self, header, message):
+    def update_status(self, header: Mapping[str, Any], message: Any) -> None:
         """Process incoming status message. Acquire lock for status dictionary before updating."""
         with self._lock:
             if self.message_box:
@@ -70,14 +70,21 @@ def update_status(self, header, message):
                 self._node_status[message["host"]] = message
                 self._node_status[message["host"]]["last_seen"] = receipt_time
 
-    def run(self):
+    def run(self) -> None:
         """A wrapper for the real _run() function to cleanly enable/disable the
         curses environment."""
         curses.wrapper(self._run)
 
     def _boxwin(
-        self, height, width, row, column, title=None, title_x=7, color_pair=None
-    ):
+        self,
+        height: int,
+        width: int,
+        row: int,
+        column: int,
+        title: str | None = None,
+        title_x: int = 7,
+        color_pair: int | None = None,
+    ) -> curses.window:
         with self._lock:
             box = curses.newwin(height, width, row, column)
             box.clear()
@@ -91,7 +98,7 @@ def _boxwin(
             box.noutrefresh()
             return curses.newwin(height - 2, width - 2, row + 1, column + 1)
 
-    def _redraw_screen(self, stdscr):
+    def _redraw_screen(self, stdscr: curses.window) -> None:
         """Redraw screen. This could be to initialize, or to redraw after resizing."""
         with self._lock:
             stdscr.clear()
@@ -105,7 +112,7 @@ def _redraw_screen(self, stdscr):
             self.message_box.scrollok(True)
             self.cards = []
 
-    def _get_card(self, number):
+    def _get_card(self, number: int) -> curses.window:
         with self._lock:
             if number < len(self.cards):
                 return self.cards[number]
@@ -123,7 +130,7 @@ def _get_card(self, number):
                 return self.cards[number]
             raise RuntimeError("Card number too high")
 
-    def _erase_card(self, number):
+    def _erase_card(self, number: int) -> None:
         """Destroy cards with this or higher number."""
         with self._lock:
             if number < (len(self.cards) - 1):
@@ -141,7 +148,7 @@ def _erase_card(self, number):
             obliterate.noutrefresh()
             del self.cards[number]
 
-    def _run(self, stdscr):
+    def _run(self, stdscr: curses.window) -> None:
         """Start the actual service monitor"""
         with self._lock:
             curses.use_default_colors()