refine type hinting

Author: scivision

pyproject.toml

@@ -42,7 +42,3 @@ line-length = 90
 
 [tool.mypy]
 files = ["src"]
-strict_optional = false
-allow_redefinition = true
-show_error_context = false
-show_column_numbers = true

src/findssh/base.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 import ipaddress as ip
 import socket
-import typing as T
+import logging
+from collections.abc import Iterable
 
 
-def get_service(b: bytes, service: str | None = None) -> str:
+def get_service(b: bytes, service: str | None = None) -> str | None:
     """
     splitlines is in case the ASCII/UTF8 response is less than 32 bytes,
     hoping server sends a \r\n
@@ -20,7 +21,7 @@ def get_service(b: bytes, service: str | None = None) -> str:
 
 def is_port_open(
     host: ip.IPv4Address, port: int, timeout: float, service: str | None = None
-) -> tuple[ip.IPv4Address, str]:
+) -> tuple[ip.IPv4Address, str] | None:
     """
     is a port open? Without coroutines.
     """
@@ -38,7 +39,8 @@ def is_port_open(
         try:
             if not (resp := s.recv(32)):
                 return None
-        except (socket.timeout, ConnectionError):
+        except (socket.timeout, ConnectionError) as err:
+            logging.debug(err)
             return None
 
     if svc_txt := get_service(resp, service):
@@ -49,7 +51,7 @@ def is_port_open(
 
 def get_hosts_seq(
     net: ip.IPv4Network, port: int, timeout: float, service: str | None = None
-) -> T.Iterable[tuple[ip.IPv4Address, str]]:
+) -> Iterable[tuple[ip.IPv4Address, str]]:
     """
     find hosts sequentially (no parallelism or concurrency)
     """

src/findssh/coro.py

@@ -35,8 +35,8 @@ async def get_hosts(
 
 
 async def waiter(
-    host: ip.IPv4Address, port: int, service: str, timeout: float
-) -> tuple[ip.IPv4Address, str]:
+    host: ip.IPv4Address, port: int, service: str | None, timeout: float
+) -> tuple[ip.IPv4Address, str] | None:
     try:
         res = await asyncio.wait_for(is_port_open(host, port, service), timeout=timeout)
     except asyncio.TimeoutError:
@@ -46,8 +46,8 @@ async def waiter(
 
 
 async def is_port_open(
-    host: ip.IPv4Address, port: int, service: str
-) -> tuple[ip.IPv4Address, str]:
+    host: ip.IPv4Address, port: int, service: str | None
+) -> tuple[ip.IPv4Address, str] | None:
     """
     https://docs.python.org/3/library/asyncio-stream.html#asyncio.open_connection
     """

src/findssh/tests/test_threadpool.py

@@ -1,6 +1,7 @@
 """
 normally we use coroutines, but for demo purposes we have threadpool too.
 """
+
 import ipaddress
 
 import findssh.threadpool

src/findssh/threadpool.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 import concurrent.futures
 import ipaddress as ip
-import typing as T
+from collections.abc import Iterable
 
 from .base import is_port_open
 
@@ -19,7 +19,7 @@ def get_hosts(
     port: int,
     timeout: float,
     service: str | None = None,
-) -> T.Iterable[tuple[ip.IPv4Address, str]]:
+) -> Iterable[tuple[ip.IPv4Address, str]]:
     """
     loops over hosts in network, one thread per address.
     This is MUCH slower than asyncio coroutines in coro.py