Public interface entirely unambiguously typed

Author: mikeshardmind

pyproject.toml

@@ -1,5 +1,5 @@
 [tool.pyright]
-include = ["async_utils"]
+include = ["src"]
 exclude = [
     "**/__pycache__",
     "build",

src/async_utils/__init__.py

@@ -14,13 +14,13 @@
 import os
 import sys
 
-vi = sys.version_info
+_vi = sys.version_info
 
 # - Check use of concurrent.futures.Future before extending this version.
 # - update `_misc._ensure_annotations.py` before extending this version.
 # - ensure `task_cache.__WrappedSignature` still works
-if (vi.major, vi.minor) > (3, 13):
-    msg = """This library is not tested for use on python versions above 3.13
+if (_vi.major, _vi.minor) > (3, 13):
+    msg: str = """This library is not tested for use on python versions above 3.13
     This library relies on a few internal details that are not safe to rely upon
     without checking this consistently.
     """

src/async_utils/bg_loop.py

@@ -37,7 +37,7 @@ def __init_subclass__(cls) -> t.Never:
     __final__ = True
 
     def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
-        self._loop = loop
+        self._loop: asyncio.AbstractEventLoop = loop
         self._futures: set[cf.Future[t.Any]] = set()
 
     def schedule[T](self, coro: CoroReturning[T], /) -> cf.Future[T]:

src/async_utils/priority_sem.py

@@ -35,8 +35,8 @@ class PriorityWaiter:
     __slots__ = ("future", "ord")
 
     def __init__(self, priority: int, ts: float, future: asyncio.Future[None], /) -> None:
-        self.ord = (priority, ts)
-        self.future = future
+        self.future: asyncio.Future[None] = future
+        self.ord: tuple[int, float] = (priority, ts)
 
     def cancelled(self) -> bool:
         return self.future.cancelled()

src/async_utils/sig_service.py

@@ -19,7 +19,7 @@
 import signal
 import socket
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from types import FrameType
 
 from . import _typings as t
@@ -32,9 +32,9 @@
 type _HANDLER = _HTC | int | signal.Handlers | None
 
 type HandleableSignals = t.Literal["SIGINT", "SIGTERM", "SIGBREAK", "SIGHUP"]
-type SignalTuple = tuple[HandleableSignals, ...]
+type SignalSequence = Sequence[HandleableSignals]
 
-default_handled: SignalTuple = "SIGINT", "SIGTERM", "SIGBREAK"
+default_handled: SignalSequence = "SIGINT", "SIGTERM", "SIGBREAK"
 
 
 class SpecialExit(enum.IntEnum):
@@ -57,11 +57,13 @@ def __init_subclass__(cls) -> t.Never:
 
     __final__ = True
 
-    def __init__(self, signals: SignalTuple = default_handled, /) -> None:
+    def __init__(self, signals: SignalSequence = default_handled, /) -> None:
         self._startup: list[StartStopCall] = []
         self._cbs: list[SignalCallback] = []
         self._joins: list[StartStopCall] = []
-        self.ss, self.cs = socket.socketpair()
+        ss, cs = socket.socketpair()
+        self.ss: socket.socket = ss
+        self.cs: socket.socket = cs
         self.ss.setblocking(False)
         self.cs.setblocking(False)
         sig_members = signal.Signals.__members__.items()