Bump mypy; fix mypy failure

.pre-commit-config.yaml

@@ -1,22 +1,22 @@
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.5
+    rev: v0.14.10
     hooks:
       - id: ruff-check
   - repo: https://github.com/psf/black-pre-commit-mirror
-    rev: 25.11.0
+    rev: 25.12.0
     hooks:
       - id: black
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.3.0
+    rev: v2.4.1
     hooks:
       -   id: codespell
           additional_dependencies:
             - tomli
           types_or: [rst, markdown]
           files: docs
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.11.2
+    rev: v1.19.1
     hooks:
       - id: mypy
         # Override default --ignore-missing-imports

continuous_integration/scripts/test_report.py

@@ -489,7 +489,7 @@ def main(argv: list[str] | None = None) -> None:
         total.groupby([total.file, total.test])
         .filter(lambda g: (g.status == "x").sum() >= args.nfails)
         .reset_index()
-        .assign(test=lambda df: df.file + "." + df.test)  # type: ignore
+        .assign(test=lambda df: df.file + "." + df.test)
         .groupby("test")
     )
     overall = {name: grouped.get_group(name) for name in grouped.groups}

distributed/client.py

@@ -3094,6 +3094,7 @@ async def _run(
             elif resp["status"] == "error":
                 # Exception raised by the remote function
                 _, exc, tb = clean_exception(**resp)
+                assert exc is not None
                 exc = exc.with_traceback(tb)
             else:
                 assert resp["status"] == "OK"

distributed/comm/inproc.py

@@ -181,7 +181,7 @@ def __init__(  # type: ignore[no-untyped-def]
         self._closed = False
 
         self._finalizer = weakref.finalize(self, self._get_finalizer())
-        self._finalizer.atexit = False
+        self._finalizer.atexit = False  # type: ignore[misc]
         self._initialized = True
 
     def _get_finalizer(self):

distributed/comm/tcp.py

@@ -178,7 +178,7 @@ def __init__(
         self._peer_addr = peer_addr
         self.stream = stream
         self._finalizer = weakref.finalize(self, self._get_finalizer())
-        self._finalizer.atexit = False
+        self._finalizer.atexit = False  # type: ignore[misc]
         self._extra: dict = {}
 
         ref = weakref.ref(self)

distributed/compatibility.py

@@ -14,7 +14,7 @@
 
 logging_names: dict[str | int, int | str] = {}
 logging_names.update(logging._levelToName)  # type: ignore
-logging_names.update(logging._nameToLevel)  # type: ignore
+logging_names.update(logging._nameToLevel)
 
 LINUX = sys.platform == "linux"
 MACOS = sys.platform == "darwin"

distributed/diagnostics/memory_sampler.py

@@ -143,7 +143,7 @@ def to_pandas(self, *, align: bool = False) -> pd.DataFrame:
             if align:
                 # convert datetime to timedelta from the first sample
                 s.index -= s.index[0]
-            ss[label] = s[~s.index.duplicated()]  # type: ignore[attr-defined]
+            ss[label] = s[~s.index.duplicated()]
 
         df = pd.DataFrame(ss)
 

distributed/itertools.py

@@ -35,7 +35,7 @@ def ffill(x: Iterable[X], xp: Iterable[X], fp: Iterable[Y], left: Y) -> Iterator
     xp_done = False
     xp1, fp1 = None, left
     for xi in x:
-        while not xp_done and (xp1 is None or xi >= xp1):  # type: ignore[unreachable]
+        while not xp_done and (xp1 is None or xi >= xp1):  # type: ignore[operator]
             fp0 = fp1
             try:
                 xp1, fp1 = next(it)

distributed/protocol/utils.py

@@ -232,7 +232,7 @@ def merge_memoryviews(mvs: Sequence[memoryview]) -> memoryview:
     base_start_addr = address_of_memoryview(base_mv)
     start_index = first_start_addr - base_start_addr
 
-    return base_mv[start_index : start_index + nbytes].cast(format)
+    return base_mv[start_index : start_index + nbytes].cast(format)  # type: ignore[call-overload]
 
 
 one_byte_carr = ctypes.c_byte * 1

distributed/scheduler.py

@@ -1073,7 +1073,7 @@ def remove_group(self, tg: TaskGroup) -> None:
                 del self._types[typename]
 
     @property
-    @_deprecated(use_instead="groups")  # type: ignore[misc]
+    @_deprecated(use_instead="groups")  # type: ignore[untyped-decorator]
     def active(self) -> Set[TaskGroup]:
         return self.groups
 
@@ -1083,7 +1083,7 @@ def groups(self) -> Set[TaskGroup]:
         return self._groups.keys()
 
     @property
-    @_deprecated(use_instead="states")  # type: ignore[misc]
+    @_deprecated(use_instead="states")  # type: ignore[untyped-decorator]
     def active_states(self) -> dict[TaskStateState, int]:
         return self.states
 
@@ -3821,7 +3821,7 @@ def __init__(
             preload_argv = dask.config.get("distributed.scheduler.preload-argv")
         self.preloads = preloading.process_preloads(
             self,
-            preload,  # type: ignore
+            preload,
             preload_argv,
         )
 

distributed/shuffle/_rechunk.py

@@ -125,7 +125,7 @@
 from dask.highlevelgraph import HighLevelGraph
 from dask.layers import Layer
 from dask.tokenize import tokenize
-from dask.typing import Key
+from dask.typing import DaskCollection, Key
 from dask.utils import parse_bytes
 
 from distributed.core import PooledRPCCall
@@ -248,7 +248,7 @@ def rechunk_p2p(
         disk=disk,
     )
     return new_da_object(
-        HighLevelGraph.from_collections(name, layer, [x]),
+        HighLevelGraph.from_collections(name, layer, [cast(DaskCollection, x)]),
         name,
         chunks,
         meta=x._meta,

distributed/shuffle/tests/test_shuffle.py

@@ -3033,7 +3033,7 @@ async def shuffle_inputs_done(self, comm: Comm, *args: Any, **kwargs: Any) -> No
                     # like a timeout while an exception that is being raised by
                     # the handler will be serialized and sent to the scheduler
                     comm.abort()
-                raise exc_type  # type: ignore
+                raise exc_type
         return await super().shuffle_inputs_done(*args, **kwargs)
 
 

distributed/tests/test_counter.py

@@ -17,7 +17,7 @@
         pytest.param(
             Digest,
             lambda x: x.size(),
-            marks=pytest.mark.skipif(not Digest, reason="no crick library"),
+            marks=pytest.mark.skipif(Digest is None, reason="no crick library"),
         ),
     ],
 )

distributed/tests/test_worker_metrics.py

@@ -33,10 +33,10 @@ def get_digests(
     d = w.digests_total if isinstance(w, Worker) else w.cumulative_worker_metrics
     digests = {
         k: v
-        for k, v in d.items()  # type: ignore
+        for k, v in d.items()
         if k
         not in {"latency", "tick-duration", "transfer-bandwidth", "transfer-duration"}
-        and (any(a in k for a in allow) or not allow)
+        and (any(isinstance(k, tuple) and a in k for a in allow) or not allow)
     }
     assert all(v >= 0 for v in digests.values()), digests
     return digests

distributed/utils.py

@@ -1396,7 +1396,7 @@ def parse_ports(port: int | str | Collection[int] | None) -> list[int] | list[No
     if isinstance(port, Collection):
         if not all(isinstance(p, int) for p in port):
             raise TypeError(port)
-        return list(port)  # type: ignore
+        return list(port)
 
     raise TypeError(port)
 

distributed/worker.py

@@ -1370,7 +1370,7 @@ async def gather(self, who_has: dict[Key, list[str]]) -> dict[Key, object]:
             return {"status": "OK"}
 
     def get_monitor_info(self, recent: bool = False, start: int = 0) -> dict[str, Any]:
-        result = dict(
+        result: dict[str, Any] = dict(
             range_query=(
                 self.monitor.recent()
                 if recent