Enable mypy's disallow_any_explicit flag.

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 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,
+        ) -> object | 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
@@ -101,7 +103,7 @@
                 and operation.args[0] == "tb_next"
             ):  # pragma: no cover
                 return tb_next
             return operation.delegate()  # Delegate is reverting to original behaviour
 
         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

@@ -253,9 +253,10 @@
 # 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:
+# Explicit "Any" is not allowed
+class AFDPollOp:  # type: ignore[misc]
     lpOverlapped: CData
-    poll_info: Any
+    poll_info: Any  # type: ignore[misc]
     waiters: AFDWaiters
     afd_group: AFDGroup
 
@@ -684,7 +685,8 @@
 
             lpOverlapped = ffi.new("LPOVERLAPPED")
 
-            poll_info: Any = ffi.new("AFD_POLL_INFO *")
+            # Explicit "Any" is not allowed
+            poll_info: Any = ffi.new("AFD_POLL_INFO *")  # type: ignore[misc]
             poll_info.Timeout = 2**63 - 1  # INT64_MAX
             poll_info.NumberOfHandles = 1
             poll_info.Exclusive = 0

src/trio/_core/_run.py

@@ -71,25 +71,26 @@
     # for some strange reason Sphinx works with outcome.Outcome, but not Outcome, in
     # start_guest_run. Same with types.FrameType in iter_await_frames
     import outcome
-    from typing_extensions import Self, TypeVar, TypeVarTuple, Unpack
+    from typing_extensions import ParamSpec, Self, TypeVar, TypeVarTuple, Unpack
 
     PosArgT = TypeVarTuple("PosArgT")
     StatusT = TypeVar("StatusT", default=None)
     StatusT_contra = TypeVar("StatusT_contra", contravariant=True, default=None)
+    PS = ParamSpec("PS")
 else:
     from typing import TypeVar
 
     StatusT = TypeVar("StatusT")
     StatusT_contra = TypeVar("StatusT_contra", contravariant=True)
+    PS = TypeVar("PS")
 
-FnT = TypeVar("FnT", bound="Callable[..., Any]")
 RetT = TypeVar("RetT")
 
 
 DEADLINE_HEAP_MIN_PRUNE_THRESHOLD: Final = 1000
 
 # Passed as a sentinel
-_NO_SEND: Final[Outcome[Any]] = cast("Outcome[Any]", object())
+_NO_SEND: Final[Outcome[object]] = cast("Outcome[object]", object())
 
 # Used to track if an exceptiongroup can be collapsed
 NONSTRICT_EXCEPTIONGROUP_NOTE = 'This is a "loose" ExceptionGroup, and may be collapsed by Trio if it only contains one exception - typically after `Cancelled` has been stripped from it. Note this has consequences for exception handling, and strict_exception_groups=True is recommended.'
@@ -102,7 +103,7 @@
 
 # Decorator to mark methods public. This does nothing by itself, but
 # trio/_tools/gen_exports.py looks for it.
-def _public(fn: FnT) -> FnT:
+def _public(fn: Callable[PS, RetT]) -> Callable[PS, RetT]:
     return fn
 
 
@@ -1172,7 +1173,11 @@
                 self._parent_waiting_in_aexit = False
                 GLOBAL_RUN_CONTEXT.runner.reschedule(self._parent_task)
 
-    def _child_finished(self, task: Task, outcome: Outcome[Any]) -> None:
+    def _child_finished(
+        self,
+        task: Task,
+        outcome: Outcome[object],
+    ) -> None:
         self._children.remove(task)
         if isinstance(outcome, Error):
             self._add_exc(outcome.error)
@@ -1278,12 +1283,13 @@
         """
         GLOBAL_RUN_CONTEXT.runner.spawn_impl(async_fn, args, self, name)
 
-    async def start(
+    # Explicit "Any" is not allowed
+    async def start(  # type: ignore[misc]
         self,
         async_fn: Callable[..., Awaitable[object]],
         *args: object,
         name: object = None,
-    ) -> Any:
+    ) -> Any | None:
         r"""Creates and initializes a child task.
 
         Like :meth:`start_soon`, but blocks until the new task has
@@ -1334,7 +1340,10 @@
                 # set strict_exception_groups = True to make sure we always unwrap
                 # *this* nursery's exceptiongroup
                 async with open_nursery(strict_exception_groups=True) as old_nursery:
-                    task_status: _TaskStatus[Any] = _TaskStatus(old_nursery, self)
+                    task_status: _TaskStatus[object | None] = _TaskStatus(
+                        old_nursery,
+                        self,
+                    )
                     thunk = functools.partial(async_fn, task_status=task_status)
                     task = GLOBAL_RUN_CONTEXT.runner.spawn_impl(
                         thunk,
@@ -1375,9 +1384,10 @@
 
 @final
 @attrs.define(eq=False, repr=False)
-class Task(metaclass=NoPublicConstructor):
+class Task(metaclass=NoPublicConstructor):  # type: ignore[misc]
     _parent_nursery: Nursery | None
-    coro: Coroutine[Any, Outcome[object], Any]
+    # Explicit "Any" is not allowed
+    coro: Coroutine[Any, Outcome[object], Any]  # type: ignore[misc]
     _runner: Runner
     name: str
     context: contextvars.Context
@@ -1395,10 +1405,11 @@
     #   tracebacks with extraneous frames.
     # - for scheduled tasks, custom_sleep_data is None
     # Tasks start out unscheduled.
-    _next_send_fn: Callable[[Any], object] | None = None
-    _next_send: Outcome[Any] | None | BaseException = None
+    # Explicit "Any" is not allowed
+    _next_send_fn: Callable[[Any], object] | None = None  # type: ignore[misc]
+    _next_send: Outcome[Any] | None | BaseException = None  # type: ignore[misc]
     _abort_func: Callable[[_core.RaiseCancelT], Abort] | None = None
-    custom_sleep_data: Any = None
+    custom_sleep_data: Any = None  # type: ignore[misc]
 
     # For introspection and nursery.start()
     _child_nurseries: list[Nursery] = attrs.Factory(list)
@@ -1466,7 +1477,8 @@
 
         """
         # Ignore static typing as we're doing lots of dynamic introspection
-        coro: Any = self.coro
+        # Explicit "Any" is not allowed
+        coro: Any = self.coro  # type: ignore[misc]
         while coro is not None:
             if hasattr(coro, "cr_frame"):
                 # A real coroutine
@@ -1611,13 +1623,16 @@
 
 
 @attrs.define(eq=False)
-class GuestState:
+# Explicit "Any" is not allowed
+class GuestState:  # type: ignore[misc]
     runner: Runner
     run_sync_soon_threadsafe: Callable[[Callable[[], object]], object]
     run_sync_soon_not_threadsafe: Callable[[Callable[[], object]], object]
-    done_callback: Callable[[Outcome[Any]], object]
+    # Explicit "Any" is not allowed
+    done_callback: Callable[[Outcome[Any]], object]  # type: ignore[misc]
     unrolled_run_gen: Generator[float, EventResult, None]
-    unrolled_run_next_send: Outcome[Any] = attrs.Factory(lambda: Value(None))
+    # Explicit "Any" is not allowed
+    unrolled_run_next_send: Outcome[Any] = attrs.Factory(lambda: Value(None))  # type: ignore[misc]
 
     def guest_tick(self) -> None:
         prev_library, sniffio_library.name = sniffio_library.name, "trio"
@@ -1662,15 +1677,17 @@
 
 
 @attrs.define(eq=False)
-class Runner:
+# Explicit "Any" is not allowed
+class Runner:  # type: ignore[misc]
     clock: Clock
     instruments: Instruments
     io_manager: TheIOManager
     ki_manager: KIManager
     strict_exception_groups: bool
 
     # Run-local values, see _local.py
-    _locals: dict[_core.RunVar[Any], Any] = attrs.Factory(dict)
+    # Explicit "Any" is not allowed
+    _locals: dict[_core.RunVar[Any], object] = attrs.Factory(dict)  # type: ignore[misc]
 
     runq: deque[Task] = attrs.Factory(deque)
     tasks: set[Task] = attrs.Factory(set)
@@ -1681,7 +1698,7 @@
     system_nursery: Nursery | None = None
     system_context: contextvars.Context = attrs.field(kw_only=True)
     main_task: Task | None = None
-    main_task_outcome: Outcome[Any] | None = None
+    main_task_outcome: Outcome[object] | None = None
 
     entry_queue: EntryQueue = attrs.Factory(EntryQueue)
     trio_token: TrioToken | None = None
@@ -1774,11 +1791,7 @@
     ################
 
     @_public
-    def reschedule(
-        self,
-        task: Task,
-        next_send: Outcome[object] = _NO_SEND,
-    ) -> None:
+    def reschedule(self, task: Task, next_send: Outcome[object] = _NO_SEND) -> None:
         """Reschedule the given task with the given
         :class:`outcome.Outcome`.
 
@@ -1889,7 +1902,7 @@
         self.reschedule(task, None)  # type: ignore[arg-type]
         return task
 
-    def task_exited(self, task: Task, outcome: Outcome[Any]) -> None:
+    def task_exited(self, task: Task, outcome: Outcome[object]) -> None:
         # break parking lots associated with the exiting task
         if task in GLOBAL_PARKING_LOT_BREAKER:
             for lot in GLOBAL_PARKING_LOT_BREAKER[task]:
@@ -2101,76 +2114,77 @@
 
     # sortedcontainers doesn't have types, and is reportedly very hard to type:
     # https://github.com/grantjenks/python-sortedcontainers/issues/68
-    waiting_for_idle: Any = attrs.Factory(SortedDict)
+    # Explicit "Any" is not allowed
+    waiting_for_idle: Any = attrs.Factory(SortedDict)  # type: ignore[misc]
 
     @_public
     async def wait_all_tasks_blocked(self, cushion: float = 0.0) -> None:
         """Block until there are no runnable tasks.
 
         This is useful in testing code when you want to give other tasks a
         chance to "settle down". The calling task is blocked, and doesn't wake
         up until all other tasks are also blocked for at least ``cushion``
         seconds. (Setting a non-zero ``cushion`` is intended to handle cases
         like two tasks talking to each other over a local socket, where we
         want to ignore the potential brief moment between a send and receive
         when all tasks are blocked.)
 
         Note that ``cushion`` is measured in *real* time, not the Trio clock
         time.
 
         If there are multiple tasks blocked in :func:`wait_all_tasks_blocked`,
         then the one with the shortest ``cushion`` is the one woken (and
         this task becoming unblocked resets the timers for the remaining
         tasks). If there are multiple tasks that have exactly the same
         ``cushion``, then all are woken.
 
         You should also consider :class:`trio.testing.Sequencer`, which
         provides a more explicit way to control execution ordering within a
         test, and will often produce more readable tests.
 
         Example:
           Here's an example of one way to test that Trio's locks are fair: we
           take the lock in the parent, start a child, wait for the child to be
           blocked waiting for the lock (!), and then check that we can't
           release and immediately re-acquire the lock::
 
              async def lock_taker(lock):
                  await lock.acquire()
                  lock.release()
 
              async def test_lock_fairness():
                  lock = trio.Lock()
                  await lock.acquire()
                  async with trio.open_nursery() as nursery:
                      nursery.start_soon(lock_taker, lock)
                      # child hasn't run yet, we have the lock
                      assert lock.locked()
                      assert lock._owner is trio.lowlevel.current_task()
                      await trio.testing.wait_all_tasks_blocked()
                      # now the child has run and is blocked on lock.acquire(), we
                      # still have the lock
                      assert lock.locked()
                      assert lock._owner is trio.lowlevel.current_task()
                      lock.release()
                      try:
                          # The child has a prior claim, so we can't have it
                          lock.acquire_nowait()
                      except trio.WouldBlock:
                          assert lock._owner is not trio.lowlevel.current_task()
                          print("PASS")
                      else:
                          print("FAIL")
 
         """
         task = current_task()
         key = (cushion, id(task))
         self.waiting_for_idle[key] = task
 
         def abort(_: _core.RaiseCancelT) -> Abort:
             del self.waiting_for_idle[key]
             return Abort.SUCCEEDED
 
         await wait_task_rescheduled(abort)
 
 
 ################################################################
@@ -2402,7 +2416,7 @@
         raise AssertionError(runner.main_task_outcome)
 
 
-def start_guest_run(
+def start_guest_run(  # type: ignore[misc]
     async_fn: Callable[..., Awaitable[RetT]],
     *args: object,
     run_sync_soon_threadsafe: Callable[[Callable[[], object]], object],
@@ -2706,7 +2720,7 @@
                 next_send_fn = task._next_send_fn
                 next_send = task._next_send
                 task._next_send_fn = task._next_send = None
-                final_outcome: Outcome[Any] | None = None
+                final_outcome: Outcome[object] | None = None
                 try:
                     # We used to unwrap the Outcome object here and send/throw
                     # its contents in directly, but it turns out that .throw()
@@ -2815,15 +2829,15 @@
 ################################################################
 
 
-class _TaskStatusIgnored(TaskStatus[Any]):
+class _TaskStatusIgnored(TaskStatus[object]):
     def __repr__(self) -> str:
         return "TASK_STATUS_IGNORED"
 
-    def started(self, value: Any = None) -> None:
+    def started(self, value: object = None) -> None:
         pass
 
 
-TASK_STATUS_IGNORED: Final[TaskStatus[Any]] = _TaskStatusIgnored()
+TASK_STATUS_IGNORED: Final[TaskStatus[object]] = _TaskStatusIgnored()
 
 
 def current_task() -> Task:

src/trio/_core/_tests/test_guest_mode.py

@@ -46,13 +46,14 @@
 #   our main
 # - final result is returned
 # - any unhandled exceptions cause an immediate crash
-def trivial_guest_run(
+# Explicit "Any" is not allowed
+def trivial_guest_run(  # type: ignore[misc]
     trio_fn: Callable[..., Awaitable[T]],
     *,
     in_host_after_start: Callable[[], None] | None = None,
     **start_guest_run_kwargs: Any,
 ) -> T:
-    todo: queue.Queue[tuple[str, Outcome[T] | Callable[..., object]]] = queue.Queue()
+    todo: queue.Queue[tuple[str, Outcome[T] | Callable[[], object]]] = queue.Queue()
 
     host_thread = threading.current_thread()
 
@@ -430,7 +431,8 @@ async def abandoned_main(in_host: InHost) -> None:
             trio.current_time()
 
 
-def aiotrio_run(
+# Explicit "Any" is not allowed
+def aiotrio_run(  # type: ignore[misc]
     trio_fn: Callable[..., Awaitable[T]],
     *,
     pass_not_threadsafe: bool = True,
@@ -555,7 +557,8 @@ async def crash_in_worker_thread_io(in_host: InHost) -> None:
             t = threading.current_thread()
             old_get_events = trio._core._run.TheIOManager.get_events
 
-            def bad_get_events(*args: Any) -> object:
+            # Explicit "Any" is not allowed
+            def bad_get_events(*args: Any) -> object:  # type: ignore[misc]
                 if threading.current_thread() is not t:
                     raise ValueError("oh no!")
                 else: