Enable mypy's disallow_any_explicit flag.

newsfragments/3121.misc.rst

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

src/trio/_core/_concat_tb.py

@@ -103,9 +103,8 @@ def controller(  # type: ignore[no-any-unimported]
                 and operation.args[0] == "tb_next"
             ) or TYPE_CHECKING:  # pragma: no cover
                 return tb_next
-            if TYPE_CHECKING:
-                raise RuntimeError("Should not be possible")
-            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/_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,14 +250,26 @@ class AFDWaiters:
     current_op: AFDPollOp | None = None
 
 
+class AFDHandle(Protocol):
+    Handle: Handle
+    Status: int
+    Events: int
+
+
+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)
-# Explicit "Any" is not allowed
-class AFDPollOp:  # type: ignore[misc]
+class AFDPollOp:
     lpOverlapped: CData
-    poll_info: Any  # type: ignore[misc]
+    poll_info: AFDPollInfo
     waiters: AFDWaiters
     afd_group: AFDGroup
 
@@ -685,8 +698,7 @@ def _refresh_afd(self, base_handle: Handle) -> None:
 
             lpOverlapped = ffi.new("LPOVERLAPPED")
 
-            # Explicit "Any" is not allowed
-            poll_info: Any = ffi.new("AFD_POLL_INFO *")  # type: ignore[misc]
+            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
@@ -699,9 +711,9 @@ def _refresh_afd(self, base_handle: Handle) -> None:
                     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/_run.py

@@ -1475,7 +1475,6 @@ def print_stack_for_task(task):
 
         """
         # Ignore static typing as we're doing lots of dynamic introspection
-        # Explicit "Any" is not allowed
         coro: Any = self.coro  # type: ignore[misc]
         while coro is not None:
             if hasattr(coro, "cr_frame"):

src/trio/_core/_tests/test_run.py

@@ -10,7 +10,7 @@
 import weakref
 from contextlib import ExitStack, contextmanager, suppress
 from math import inf, nan
-from typing import TYPE_CHECKING, Any, NoReturn, TypeVar, cast
+from typing import TYPE_CHECKING, NoReturn, TypeVar
 
 import outcome
 import pytest
@@ -2295,7 +2295,8 @@ async def detachable_coroutine(
         await sleep(0)
         nonlocal task, pdco_outcome
         task = _core.current_task()
-        pdco_outcome = await outcome.acapture(
+        # `No overload variant of "acapture" matches argument types "Callable[[Outcome[object]], Coroutine[Any, Any, object]]", "Outcome[None]"`
+        pdco_outcome = await outcome.acapture(  # type: ignore[call-overload]
             _core.permanently_detach_coroutine_object,
             task_outcome,
         )
@@ -2308,18 +2309,11 @@ async def detachable_coroutine(
     # is still iterable. At that point anything can be sent into the coroutine, so the .coro type
     # is wrong.
     assert pdco_outcome is None
-    # Explicit "Any" is not allowed
-    assert (
-        not_none(task).coro.send(
-            cast(Any, "be free!"),  # type: ignore[misc]
-        )
-        == "I'm free!"
-    )
+    # `Argument 1 to "send" of "Coroutine" has incompatible type "str"; expected "Outcome[object]"`
+    assert not_none(task).coro.send("be free!") == "I'm free!"  # type: ignore[arg-type]
     assert pdco_outcome == outcome.Value("be free!")
     with pytest.raises(StopIteration):
-        not_none(task).coro.send(
-            cast(Any, None),  # type: ignore[misc]
-        )
+        not_none(task).coro.send(None)  # type: ignore[arg-type]
 
     # Check the exception paths too
     task = None
@@ -2332,7 +2326,7 @@ async def detachable_coroutine(
     assert not_none(task).coro.throw(throw_in) == "uh oh"
     assert pdco_outcome == outcome.Error(throw_in)
     with pytest.raises(StopIteration):
-        task.coro.send(cast(Any, None))
+        task.coro.send(None)
 
     async def bad_detach() -> None:
         async with _core.open_nursery():
@@ -2384,25 +2378,10 @@ def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:  # pragma: no cover
         await wait_all_tasks_blocked()
 
         # Okay, it's detached. Here's our coroutine runner:
-        # Explicit "Any" is not allowed
-        assert (
-            not_none(task).coro.send(
-                cast(Any, "not trio!"),  # type: ignore[misc]
-            )
-            == 1
-        )
-        assert (
-            not_none(task).coro.send(
-                cast(Any, None),  # type: ignore[misc]
-            )
-            == 2
-        )
-        assert (
-            not_none(task).coro.send(
-                cast(Any, None),  # type: ignore[misc]
-            )
-            == "byebye"
-        )
+        # `Argument 1 to "send" of "Coroutine" has incompatible type "str"; expected "Outcome[object]"`
+        assert not_none(task).coro.send("not trio!") == 1  # type: ignore[arg-type]
+        assert not_none(task).coro.send(None) == 2  # type: ignore[arg-type]
+        assert not_none(task).coro.send(None) == "byebye"  # type: ignore[arg-type]
 
         # Now it's been reattached, and we can leave the nursery
 
@@ -2432,9 +2411,8 @@ def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
         await wait_all_tasks_blocked()
         assert task is not None
         nursery.cancel_scope.cancel()
-        task.coro.send(
-            cast(Any, None),  # type: ignore[misc]
-        )
+        # `Argument 1 to "send" of "Coroutine" has incompatible type "None"; expected "Outcome[object]"`
+        task.coro.send(None)  # type: ignore[arg-type]
 
     assert abort_fn_called
 

src/trio/_core/_traps.py

@@ -210,10 +210,9 @@ def abort(inner_raise_cancel):
     return (await _async_yield(WaitTaskRescheduled(abort_func))).unwrap()
 
 
-# Explicit "Any" is not allowed
-async def permanently_detach_coroutine_object(  # type: ignore[misc]
+async def permanently_detach_coroutine_object(
     final_outcome: outcome.Outcome[object],
-) -> Any:
+) -> object:
     """Permanently detach the current task from the Trio scheduler.
 
     Normally, a Trio task doesn't exit until its coroutine object exits. When

src/trio/_socket.py

@@ -750,8 +750,7 @@ async def sendto(
         __address: tuple[object, ...] | str | Buffer,
     ) -> int: ...
 
-    # Explicit "Any" is not allowed
-    async def sendto(self, *args: Any) -> int:  # type: ignore[misc]
+    async def sendto(self, *args: object) -> int:
         raise NotImplementedError
 
     if sys.platform != "win32" or (