Add types to some internal modules (_util, _deprecate, _ki)

.coveragerc

@@ -21,11 +21,14 @@ exclude_lines =
   abc.abstractmethod
   if TYPE_CHECKING:
   if _t.TYPE_CHECKING:
+  if t.TYPE_CHECKING:
   @overload
 
 partial_branches =
   pragma: no branch
   if not TYPE_CHECKING:
   if not _t.TYPE_CHECKING:
+  if not t.TYPE_CHECKING:
   if .* or not TYPE_CHECKING:
   if .* or not _t.TYPE_CHECKING:
+  if .* or not t.TYPE_CHECKING:

pyproject.toml

@@ -47,26 +47,36 @@ disallow_untyped_defs = false
 # DO NOT use `ignore_errors`; it doesn't apply
 # downstream and users have to deal with them.
 [[tool.mypy.overrides]]
+# Fully typed, enable stricter checks
 module = [
-    "trio._socket",
+    "trio._abc",
     "trio._core._local",
-    "trio._sync",
+    "trio._deprecate",
+    "trio._dtls",
     "trio._file_io",
+    "trio._ki",
+    "trio._socket",
+    "trio._sync",
+    "trio._util",
 ]
 disallow_incomplete_defs = true
 disallow_untyped_defs = true
+disallow_untyped_decorators = true
 disallow_any_generics = true
 disallow_any_decorated = true
+disallow_any_unimported = true
 disallow_subclassing_any = true
 
 [[tool.mypy.overrides]]
+# Needs to use Any due to some complex introspection.
 module = [
     "trio._path",
 ]
 disallow_incomplete_defs = true
 disallow_untyped_defs = true
 #disallow_any_generics = true
 #disallow_any_decorated = true
+disallow_any_unimported = true
 disallow_subclassing_any = true
 
 [tool.pytest.ini_options]

trio/__init__.py

@@ -1,5 +1,6 @@
 """Trio - A friendly Python library for async concurrency and I/O
 """
+from __future__ import annotations
 
 # General layout:
 #
@@ -16,7 +17,7 @@
 # Uses `from x import y as y` for compatibility with `pyright --verifytypes` (#2625)
 
 # must be imported early to avoid circular import
-from ._core import TASK_STATUS_IGNORED as TASK_STATUS_IGNORED  # isort: skip
+from ._core import TASK_STATUS_IGNORED as TASK_STATUS_IGNORED  # isort: split
 
 # Submodules imported by default
 from . import abc, from_thread, lowlevel, socket, to_thread
@@ -117,7 +118,7 @@
 
 _deprecate.enable_attribute_deprecations(__name__)
 
-__deprecated_attributes__ = {
+__deprecated_attributes__: dict[str, _deprecate.DeprecatedAttribute] = {
     "open_process": _deprecate.DeprecatedAttribute(
         value=lowlevel.open_process,
         version="0.20.0",

trio/_core/_ki.py

@@ -3,17 +3,21 @@
 import inspect
 import signal
 import sys
+import types
+from collections.abc import Callable
 from functools import wraps
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final, TypeVar
 
 import attr
 
 from .._util import is_main_thread
 
+RetT = TypeVar("RetT")
+
 if TYPE_CHECKING:
-    from typing import Any, Callable, TypeVar
+    from typing_extensions import ParamSpec, TypeGuard
 
-    F = TypeVar("F", bound=Callable[..., Any])
+    ArgsT = ParamSpec("ArgsT")
 
 # In ordinary single-threaded Python code, when you hit control-C, it raises
 # an exception and automatically does all the regular unwinding stuff.
@@ -80,22 +84,22 @@
 # We use this special string as a unique key into the frame locals dictionary.
 # The @ ensures it is not a valid identifier and can't clash with any possible
 # real local name. See: https://github.com/python-trio/trio/issues/469
-LOCALS_KEY_KI_PROTECTION_ENABLED = "@TRIO_KI_PROTECTION_ENABLED"
+LOCALS_KEY_KI_PROTECTION_ENABLED: Final = "@TRIO_KI_PROTECTION_ENABLED"
 
 
 # NB: according to the signal.signal docs, 'frame' can be None on entry to
 # this function:
-def ki_protection_enabled(frame):
+def ki_protection_enabled(frame: types.FrameType | None) -> bool:
     while frame is not None:
         if LOCALS_KEY_KI_PROTECTION_ENABLED in frame.f_locals:
-            return frame.f_locals[LOCALS_KEY_KI_PROTECTION_ENABLED]
+            return bool(frame.f_locals[LOCALS_KEY_KI_PROTECTION_ENABLED])
         if frame.f_code.co_name == "__del__":
             return True
         frame = frame.f_back
     return True
 
 
-def currently_ki_protected():
+def currently_ki_protected() -> bool:
     r"""Check whether the calling code has :exc:`KeyboardInterrupt` protection
     enabled.
 
@@ -115,29 +119,35 @@ def currently_ki_protected():
 # functions decorated @async_generator are given this magic property that's a
 # reference to the object itself
 # see python-trio/async_generator/async_generator/_impl.py
-def legacy_isasyncgenfunction(obj):
+def legacy_isasyncgenfunction(
+    obj: object,
+) -> TypeGuard[Callable[..., types.AsyncGeneratorType]]:
     return getattr(obj, "_async_gen_function", None) == id(obj)
 
 
-def _ki_protection_decorator(enabled):
-    def decorator(fn):
+def _ki_protection_decorator(
+    enabled: bool,
+) -> Callable[[Callable[ArgsT, RetT]], Callable[ArgsT, RetT]]:
+    # The "ignore[return-value]" below is because the inspect functions cast away the
+    # original return type of fn, making it just CoroutineType[Any, Any, Any] etc.
+    def decorator(fn: Callable[ArgsT, RetT]) -> Callable[ArgsT, RetT]:
         # In some version of Python, isgeneratorfunction returns true for
         # coroutine functions, so we have to check for coroutine functions
         # first.
         if inspect.iscoroutinefunction(fn):
 
             @wraps(fn)
-            def wrapper(*args, **kwargs):
+            def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
                 # See the comment for regular generators below
                 coro = fn(*args, **kwargs)
                 coro.cr_frame.f_locals[LOCALS_KEY_KI_PROTECTION_ENABLED] = enabled
-                return coro
+                return coro  # type: ignore[return-value]
 
             return wrapper
         elif inspect.isgeneratorfunction(fn):
 
             @wraps(fn)
-            def wrapper(*args, **kwargs):
+            def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
                 # It's important that we inject this directly into the
                 # generator's locals, as opposed to setting it here and then
                 # doing 'yield from'. The reason is, if a generator is
@@ -148,23 +158,23 @@ def wrapper(*args, **kwargs):
                 #     https://bugs.python.org/issue29590
                 gen = fn(*args, **kwargs)
                 gen.gi_frame.f_locals[LOCALS_KEY_KI_PROTECTION_ENABLED] = enabled
-                return gen
+                return gen  # type: ignore[return-value]
 
             return wrapper
         elif inspect.isasyncgenfunction(fn) or legacy_isasyncgenfunction(fn):
 
             @wraps(fn)
-            def wrapper(*args, **kwargs):
+            def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
                 # See the comment for regular generators above
                 agen = fn(*args, **kwargs)
                 agen.ag_frame.f_locals[LOCALS_KEY_KI_PROTECTION_ENABLED] = enabled
-                return agen
+                return agen  # type: ignore[return-value]
 
             return wrapper
         else:
 
             @wraps(fn)
-            def wrapper(*args, **kwargs):
+            def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
                 locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = enabled
                 return fn(*args, **kwargs)
 
@@ -173,26 +183,34 @@ def wrapper(*args, **kwargs):
     return decorator
 
 
-enable_ki_protection: Callable[[F], F] = _ki_protection_decorator(True)
+enable_ki_protection: Callable[
+    [Callable[ArgsT, RetT]], Callable[ArgsT, RetT]
+] = _ki_protection_decorator(True)
 enable_ki_protection.__name__ = "enable_ki_protection"
 
-disable_ki_protection: Callable[[F], F] = _ki_protection_decorator(False)
+disable_ki_protection: Callable[
+    [Callable[ArgsT, RetT]], Callable[ArgsT, RetT]
+] = _ki_protection_decorator(False)
 disable_ki_protection.__name__ = "disable_ki_protection"
 
 
 @attr.s
 class KIManager:
     handler = attr.ib(default=None)
 
-    def install(self, deliver_cb, restrict_keyboard_interrupt_to_checkpoints):
+    def install(
+        self,
+        deliver_cb: Callable[[], object],
+        restrict_keyboard_interrupt_to_checkpoints: bool,
+    ) -> None:
         assert self.handler is None
         if (
             not is_main_thread()
             or signal.getsignal(signal.SIGINT) != signal.default_int_handler
         ):
             return
 
-        def handler(signum, frame):
+        def handler(signum: int, frame: types.FrameType | None) -> None:
             assert signum == signal.SIGINT
             protection_enabled = ki_protection_enabled(frame)
             if protection_enabled or restrict_keyboard_interrupt_to_checkpoints:

trio/_deprecate.py

@@ -1,10 +1,21 @@
+from __future__ import annotations
+
 import sys
 import warnings
+from collections.abc import Callable
 from functools import wraps
 from types import ModuleType
+from typing import TYPE_CHECKING, ClassVar, TypeVar
 
 import attr
 
+if TYPE_CHECKING:
+    from typing_extensions import ParamSpec
+
+    ArgsT = ParamSpec("ArgsT")
+
+RetT = TypeVar("RetT")
+
 
 # We want our warnings to be visible by default (at least for now), but we
 # also want it to be possible to override that using the -W switch. AFAICT
@@ -29,17 +40,24 @@ class TrioDeprecationWarning(FutureWarning):
     """
 
 
-def _url_for_issue(issue):
+def _url_for_issue(issue: int) -> str:
     return f"https://github.com/python-trio/trio/issues/{issue}"
 
 
-def _stringify(thing):
+def _stringify(thing: object) -> str:
     if hasattr(thing, "__module__") and hasattr(thing, "__qualname__"):
         return f"{thing.__module__}.{thing.__qualname__}"
     return str(thing)
 
 
-def warn_deprecated(thing, version, *, issue, instead, stacklevel=2):
+def warn_deprecated(
+    thing: object,
+    version: str,
+    *,
+    issue: int | None,
+    instead: object,
+    stacklevel: int = 2,
+) -> None:
     stacklevel += 1
     msg = f"{_stringify(thing)} is deprecated since Trio {version}"
     if instead is None:
@@ -53,12 +71,14 @@ def warn_deprecated(thing, version, *, issue, instead, stacklevel=2):
 
 # @deprecated("0.2.0", issue=..., instead=...)
 # def ...
-def deprecated(version, *, thing=None, issue, instead):
-    def do_wrap(fn):
+def deprecated(
+    version: str, *, thing: object = None, issue: int | None, instead: object
+) -> Callable[[Callable[ArgsT, RetT]], Callable[ArgsT, RetT]]:
+    def do_wrap(fn: Callable[ArgsT, RetT]) -> Callable[ArgsT, RetT]:
         nonlocal thing
 
         @wraps(fn)
-        def wrapper(*args, **kwargs):
+        def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
             warn_deprecated(thing, version, instead=instead, issue=issue)
             return fn(*args, **kwargs)
 
@@ -87,11 +107,17 @@ def wrapper(*args, **kwargs):
     return do_wrap
 
 
-def deprecated_alias(old_qualname, new_fn, version, *, issue):
+def deprecated_alias(
+    old_qualname: str,
+    new_fn: Callable[ArgsT, RetT],
+    version: str,
+    *,
+    issue: int | None,
+) -> Callable[ArgsT, RetT]:
     @deprecated(version, issue=issue, instead=new_fn)
     @wraps(new_fn, assigned=("__module__", "__annotations__"))
-    def wrapper(*args, **kwargs):
-        "Deprecated alias."
+    def wrapper(*args: ArgsT.args, **kwargs: ArgsT.kwargs) -> RetT:
+        """Deprecated alias."""
         return new_fn(*args, **kwargs)
 
     wrapper.__qualname__ = old_qualname
@@ -101,16 +127,18 @@ def wrapper(*args, **kwargs):
 
 @attr.s(frozen=True)
 class DeprecatedAttribute:
-    _not_set = object()
+    _not_set: ClassVar[object] = object()
 
-    value = attr.ib()
-    version = attr.ib()
-    issue = attr.ib()
-    instead = attr.ib(default=_not_set)
+    value: object = attr.ib()
+    version: str = attr.ib()
+    issue: int | None = attr.ib()
+    instead: object = attr.ib(default=_not_set)
 
 
 class _ModuleWithDeprecations(ModuleType):
-    def __getattr__(self, name):
+    __deprecated_attributes__: dict[str, DeprecatedAttribute]
+
+    def __getattr__(self, name: str) -> object:
         if name in self.__deprecated_attributes__:
             info = self.__deprecated_attributes__[name]
             instead = info.instead
@@ -124,9 +152,10 @@ def __getattr__(self, name):
         raise AttributeError(msg.format(self.__name__, name))
 
 
-def enable_attribute_deprecations(module_name):
+def enable_attribute_deprecations(module_name: str) -> None:
     module = sys.modules[module_name]
     module.__class__ = _ModuleWithDeprecations
+    assert isinstance(module, _ModuleWithDeprecations)
     # Make sure that this is always defined so that
     # _ModuleWithDeprecations.__getattr__ can access it without jumping
     # through hoops or risking infinite recursion.

trio/_tests/test_exports.py

@@ -1,3 +1,5 @@
+import __future__  # Regular import, not special!
+
 import enum
 import functools
 import importlib
@@ -107,6 +109,11 @@ def no_underscores(symbols):
     if modname == "trio":
         runtime_names.discard("tests")
 
+    # Ignore any __future__ feature objects, if imported under that name.
+    for name in __future__.all_feature_names:
+        if getattr(module, name, None) is getattr(__future__, name):
+            runtime_names.remove(name)
+
     if tool in ("mypy", "pyright_verifytypes"):
         # create py.typed file
         py_typed_path = Path(trio.__file__).parent / "py.typed"