More type linting

Author: ndevenish

src/workflows/services/sample_consumer.py

@@ -19,7 +19,7 @@ class SampleConsumer(CommonService):
 
     def initializing(self):
         """Subscribe to a channel."""
-        self._transport.subscribe("transient.destination", self.consume_message)
+        self.transport.subscribe("transient.destination", self.consume_message)
 
     def consume_message(self, header, message):
         """Consume a message"""

src/workflows/services/sample_pipethrough.py

@@ -21,7 +21,7 @@ class SamplePipethrough(CommonService):
     def initializing(self):
         """Subscribe to a channel."""
         workflows.recipe.wrap_subscribe(
-            self._transport,
+            self.transport,
             "transient.destination",
             self.process,
         )

src/workflows/services/sample_producer.py

@@ -30,7 +30,7 @@ def create_message(self):
         """Create and send a unique message for this service."""
         self.counter += 1
         self.log.info("Sending message #%d", self.counter)
-        self._transport.send(
+        self.transport.send(
             "transient.destination",
             "Message #%d\n++++++++Produced @%10.3f ms"
             % (self.counter, (time.time() % 1000) * 1000),

src/workflows/services/sample_transaction.py

@@ -18,7 +18,7 @@ class SampleTxn(CommonService):
 
     def initializing(self):
         """Subscribe to a channel. Received messages must be acknowledged."""
-        self.subid = self._transport.subscribe(
+        self.subid = self.transport.subscribe(
             "transient.transaction",
             self.receive_message,
             acknowledgement=True,
@@ -41,29 +41,29 @@ def receive_message(self, header, message):
         self.log.info("MsgID: {}".format(header["message-id"]))
         assert header["message-id"]
 
-        txn = self._transport.transaction_begin()
+        txn = self.transport.transaction_begin()
         self.log.info(f" 1. Txn: {txn}")
         if self.crashpoint():
-            self._transport.transaction_abort(txn)
+            self.transport.transaction_abort(txn)
             self.log.info("---  Abort  ---")
             return
 
-        self._transport.ack(header["message-id"], self.subid, transaction=txn)
+        self.transport.ack(header["message-id"], self.subid, transaction=txn)
         self.log.info(" 2. Ack")
         if self.crashpoint():
-            self._transport.transaction_abort(txn)
+            self.transport.transaction_abort(txn)
             self.log.info("---  Abort  ---")
             return
 
-        self._transport.send("transient.destination", message, transaction=txn)
+        self.transport.send("transient.destination", message, transaction=txn)
         self.log.info(" 3. Send")
 
         if self.crashpoint():
-            self._transport.transaction_abort(txn)
+            self.transport.transaction_abort(txn)
             self.log.info("---  Abort  ---")
             return
 
-        self._transport.transaction_commit(txn)
+        self.transport.transaction_commit(txn)
         self.log.info(" 4. Commit")
         self.log.info("===  Done   ===")
 
@@ -88,7 +88,7 @@ def initializing(self):
     def create_message(self):
         """Create and send a unique message for this service."""
         self.counter += 1
-        self._transport.send(
+        self.transport.send(
             "transient.transaction",
             "TXMessage #%d\n++++++++Produced@ %f"
             % (self.counter, (time.time() % 1000) * 1000),

src/workflows/transport/offline_transport.py

@@ -28,7 +28,7 @@ class OfflineTransport(CommonTransport):
     config: dict[Any, Any] = {}
 
     def __init__(
-        self, middleware: list[type[middleware.BaseTransportMiddleware]] | None = None
+        self, middleware: list[middleware.BaseTransportMiddleware] | None = None
     ):
         self._connected = False
         super().__init__(middleware=middleware)
@@ -78,10 +78,8 @@ def _subscribe_broadcast(self, sub_id, channel, callback, **kwargs):
             f"subscription ID {sub_id}, callback function {callback}, further keywords: {kwargs}",
         )
 
-    def _unsubscribe(self, subscription, **kwargs):
-        self._output(
-            f"Ending subscription #{subscription}", f"further keywords: {kwargs}"
-        )
+    def _unsubscribe(self, sub_id: int, **kwargs):
+        self._output(f"Ending subscription #{sub_id}", f"further keywords: {kwargs}")
 
     def _send(
         self, destination, message, headers=None, delay=None, expiration=None, **kwargs

src/workflows/transport/pika_transport.py

@@ -15,7 +15,9 @@
 from enum import Enum, auto
 from typing import Any
 
+import pika.channel
 import pika.exceptions
+import pika.spec
 from bidict import bidict
 from pika.adapters.blocking_connection import BlockingChannel
 
@@ -63,7 +65,7 @@ class PikaTransport(CommonTransport):
     config: dict[Any, Any] = {}
 
     def __init__(
-        self, middleware: list[type[middleware.BaseTransportMiddleware]] | None = None
+        self, middleware: list[middleware.BaseTransportMiddleware] | None = None
     ):
         self._channel = None
         self._conn = None
@@ -242,17 +244,15 @@ def set_parameter(option, opt, value, parser):
         )
 
     def _generate_connection_parameters(self) -> list[pika.ConnectionParameters]:
-        username = self.config.get("--rabbit-user", self.defaults.get("--rabbit-user"))
-        password = self.config.get("--rabbit-pass", self.defaults.get("--rabbit-pass"))
+        username = self.config.get("--rabbit-user", self.defaults["--rabbit-user"])
+        password = self.config.get("--rabbit-pass", self.defaults["--rabbit-pass"])
         credentials = pika.PlainCredentials(username, password)
 
-        host_string = self.config.get(
-            "--rabbit-host", self.defaults.get("--rabbit-host")
-        )
+        host_string = self.config.get("--rabbit-host", self.defaults["--rabbit-host"])
         port_string = str(
-            self.config.get("--rabbit-port", self.defaults.get("--rabbit-port"))
+            self.config.get("--rabbit-port", self.defaults["--rabbit-port"])
         )
-        vhost = self.config.get("--rabbit-vhost", self.defaults.get("--rabbit-vhost"))
+        vhost = self.config.get("--rabbit-vhost", self.defaults["--rabbit-vhost"])
         if "," in host_string:
             host = host_string.split(",")
         else: