Enable mypy's disallow_any_explicit flag.

newsfragments/3121.misc.rst

@@ -0,0 +1 @@
+Improve type annotations in several places by removing `Any` usage.

pyproject.toml

@@ -183,6 +183,7 @@ warn_return_any = true
 
 # Avoid subtle backsliding
 disallow_any_decorated = true
+disallow_any_explicit = true
 disallow_any_generics = true
 disallow_any_unimported = true
 disallow_incomplete_defs = true

src/trio/_core/_concat_tb.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from types import TracebackType
-from typing import Any, ClassVar, cast
+from typing import TYPE_CHECKING, ClassVar, cast
 
 ################################################################
 # concat_tb
@@ -86,7 +86,9 @@
     def copy_tb(base_tb: TracebackType, tb_next: TracebackType | None) -> TracebackType:
         # tputil.ProxyOperation is PyPy-only, and there's no way to specify
         # cpython/pypy in current type checkers.
-        def controller(operation: tputil.ProxyOperation) -> Any | None:  # type: ignore[no-any-unimported]
+        def controller(  # type: ignore[no-any-unimported]
+            operation: tputil.ProxyOperation,
+        ) -> TracebackType | None:
             # Rationale for pragma: I looked fairly carefully and tried a few
             # things, and AFAICT it's not actually possible to get any
             # 'opname' that isn't __getattr__ or __getattribute__. So there's
@@ -99,9 +101,10 @@
                     "__getattr__",
                 }
                 and operation.args[0] == "tb_next"
-            ):  # pragma: no cover
+            ) or TYPE_CHECKING:  # pragma: no cover
                 return tb_next
-            return operation.delegate()  # Delegate is reverting to original behaviour
+            # Delegate is reverting to original behaviour
+            return operation.delegate()  # type: ignore[no-any-return]
 
         return cast(
             TracebackType,

src/trio/_core/_entry_queue.py

@@ -16,7 +16,8 @@
 
     PosArgsT = TypeVarTuple("PosArgsT")
 
-Function = Callable[..., object]
+# Explicit "Any" is not allowed
+Function = Callable[..., object]  # type: ignore[misc]
 Job = tuple[Function, tuple[object, ...]]
 
 

src/trio/_core/_instrumentation.py

@@ -3,20 +3,22 @@
 import logging
 import types
 from collections.abc import Callable, Sequence
-from typing import Any, TypeVar
+from typing import TypeVar
 
 from .._abc import Instrument
 
 # Used to log exceptions in instruments
 INSTRUMENT_LOGGER = logging.getLogger("trio.abc.Instrument")
 
 
-F = TypeVar("F", bound=Callable[..., Any])
+# Explicit "Any" is not allowed
+F = TypeVar("F", bound=Callable[..., object])  # type: ignore[misc]
 
 
 # Decorator to mark methods public. This does nothing by itself, but
 # trio/_tools/gen_exports.py looks for it.
-def _public(fn: F) -> F:
+# Explicit "Any" is not allowed
+def _public(fn: F) -> F:  # type: ignore[misc]
     return fn
 
 
@@ -89,7 +91,11 @@ def remove_instrument(self, instrument: Instrument) -> None:
                 if not instruments:
                     del self[hookname]
 
-    def call(self, hookname: str, *args: Any) -> None:
+    def call(
+        self,
+        hookname: str,
+        *args: object,
+    ) -> None:
         """Call hookname(*args) on each applicable instrument.
 
         You must first check whether there are any instruments installed for

src/trio/_core/_io_windows.py

@@ -7,8 +7,8 @@
 from contextlib import contextmanager
 from typing import (
     TYPE_CHECKING,
-    Any,
     Literal,
+    Protocol,
     TypeVar,
     cast,
 )
@@ -24,6 +24,7 @@
     AFDPollFlags,
     CData,
     CompletionModes,
+    CType,
     ErrorCodes,
     FileFlags,
     Handle,
@@ -249,13 +250,28 @@
     current_op: AFDPollOp | None = None
 
 
+# Just used for internal type checking.
+class _AFDHandle(Protocol):
+    Handle: Handle
+    Status: int
+    Events: int
+
+
+# Just used for internal type checking.
+class _AFDPollInfo(Protocol):
+    Timeout: int
+    NumberOfHandles: int
+    Exclusive: int
+    Handles: list[_AFDHandle]
+
+
 # We also need to bundle up all the info for a single op into a standalone
 # object, because we need to keep all these objects alive until the operation
 # finishes, even if we're throwing it away.
 @attrs.frozen(eq=False)
 class AFDPollOp:
     lpOverlapped: CData
-    poll_info: Any
+    poll_info: _AFDPollInfo
     waiters: AFDWaiters
     afd_group: AFDGroup
 
@@ -684,7 +700,7 @@
 
             lpOverlapped = ffi.new("LPOVERLAPPED")
 
-            poll_info: Any = ffi.new("AFD_POLL_INFO *")
+            poll_info = cast(_AFDPollInfo, ffi.new("AFD_POLL_INFO *"))
             poll_info.Timeout = 2**63 - 1  # INT64_MAX
             poll_info.NumberOfHandles = 1
             poll_info.Exclusive = 0
@@ -697,9 +713,9 @@
                     kernel32.DeviceIoControl(
                         afd_group.handle,
                         IoControlCodes.IOCTL_AFD_POLL,
-                        poll_info,
+                        cast(CType, poll_info),
                         ffi.sizeof("AFD_POLL_INFO"),
-                        poll_info,
+                        cast(CType, poll_info),
                         ffi.sizeof("AFD_POLL_INFO"),
                         ffi.NULL,
                         lpOverlapped,

src/trio/_core/_ki.py

@@ -4,7 +4,7 @@
 import sys
 import types
 import weakref
-from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar
+from typing import TYPE_CHECKING, Generic, Protocol, TypeVar
 
 import attrs
 
@@ -85,7 +85,12 @@ class _IdRef(weakref.ref[_T]):
     __slots__ = ("_hash",)
     _hash: int
 
-    def __new__(cls, ob: _T, callback: Callable[[Self], Any] | None = None, /) -> Self:
+    def __new__(
+        cls,
+        ob: _T,
+        callback: Callable[[Self], object] | None = None,
+        /,
+    ) -> Self:
         self: Self = weakref.ref.__new__(cls, ob, callback)
         self._hash = object.__hash__(ob)
         return self