Finish type hinting

Author: AlanCoding

dispatcher/brokers/pg_notify.py

@@ -100,6 +100,9 @@ def get_publish_channel(self, channel: Optional[str] = None) -> str:
 
         raise ValueError('Could not determine a channel to use publish to from settings or PGNotify config')
 
+    def __str__(self) -> str:
+        return 'pg_notify-broker'
+
     # --- asyncio connection methods ---
 
     async def aget_connection(self) -> psycopg.AsyncConnection:

dispatcher/brokers/socket.py

@@ -21,13 +21,13 @@ def __init__(self, client_id: int, reader: asyncio.StreamReader, writer: asyncio
         self.yield_clear = asyncio.Event()
         self.replies_to_send: list = []
 
-    def write(self, message) -> None:
+    def write(self, message: str, /) -> None:
         self.writer.write((message + '\n').encode())
 
-    def queue_reply(self, reply: str) -> None:
+    def queue_reply(self, reply: str, /) -> None:
         self.replies_to_send.append(reply)
 
-    async def send_replies(self):
+    async def send_replies(self) -> None:
         for reply in self.replies_to_send.copy():
             logger.info(f'Sending reply to client_id={self.client_id} len={len(reply)}')
             self.write(reply)
@@ -54,8 +54,8 @@ def __init__(self, socket_path: str) -> None:
         self.sock: Optional[socket.socket] = None  # for synchronous clients
         self.incoming_queue: asyncio.Queue = asyncio.Queue()
 
-    def __str__(self):
-        return f'socket-producer-{self.socket_path}'
+    def __str__(self) -> str:
+        return f'socket-broker-{self.socket_path}'
 
     async def _add_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
         client = Client(self.client_ct, reader, writer)
@@ -190,10 +190,11 @@ def process_notify(
         finally:
             self.sock = None
 
-    def _publish_from_sock(self, sock, message) -> None:
+    def _publish_from_sock(self, sock: socket.socket, message: str) -> None:
         sock.sendall((message + "\n").encode())
 
-    def publish_message(self, channel=None, message=None) -> None:
+    def publish_message(self, channel: Optional[str] = None, message: Optional[str] = None) -> None:
+        assert isinstance(message, str)
         if self.sock:
             logger.info(f'Publishing socket message len={len(message)} via existing connection')
             self._publish_from_sock(self.sock, message)

dispatcher/producers/brokered.py

@@ -15,7 +15,7 @@ def __init__(self, broker: Broker) -> None:
         self.dispatcher: Optional[DispatcherMain] = None
         super().__init__()
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f'brokered-producer-{self.broker}'
 
     async def start_producing(self, dispatcher: DispatcherMain) -> None: