Bump mypy to 1.16.1

Author: ianthomas23

.pre-commit-config.yaml

@@ -40,7 +40,7 @@ repos:
         types_or: [yaml, html, json]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: "v1.8.0"
+    rev: "v1.16.1"
     hooks:
       - id: mypy
         files: ipykernel

ipykernel/comm/comm.py

@@ -49,7 +49,7 @@ def publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys):
 class Comm(BaseComm, traitlets.config.LoggingConfigurable):
     """Class for communicating between a Frontend and a Kernel"""
 
-    kernel = Instance("ipykernel.kernelbase.Kernel", allow_none=True)  # type:ignore[assignment]
+    kernel = Instance("ipykernel.kernelbase.Kernel", allow_none=True)
     comm_id = Unicode()
     primary = Bool(True, help="Am I the primary or secondary Comm?")
 

ipykernel/inprocess/blocking.py

@@ -69,6 +69,7 @@ def call_handlers(self, msg):
             _raw_input = self.client.kernel._sys_raw_input
             prompt = msg["content"]["prompt"]
             print(prompt, end="", file=sys.__stdout__)
+            assert sys.__stdout__ is not None
             sys.__stdout__.flush()
             self.client.input(_raw_input())
 
@@ -77,9 +78,9 @@ class BlockingInProcessKernelClient(InProcessKernelClient):
     """A blocking in-process kernel client."""
 
     # The classes to use for the various channels.
-    shell_channel_class = Type(BlockingInProcessChannel)  # type:ignore[arg-type]
-    iopub_channel_class = Type(BlockingInProcessChannel)  # type:ignore[arg-type]
-    stdin_channel_class = Type(BlockingInProcessStdInChannel)  # type:ignore[arg-type]
+    shell_channel_class = Type(BlockingInProcessChannel)
+    iopub_channel_class = Type(BlockingInProcessChannel)
+    stdin_channel_class = Type(BlockingInProcessStdInChannel)
 
     def wait_for_ready(self):
         """Wait for kernel info reply on shell channel."""

ipykernel/inprocess/client.py

@@ -12,6 +12,7 @@
 # -----------------------------------------------------------------------------
 
 import asyncio
+from typing import Any
 
 from jupyter_client.client import KernelClient
 from jupyter_client.clientabc import KernelClientABC
@@ -39,11 +40,11 @@ class InProcessKernelClient(KernelClient):
     """
 
     # The classes to use for the various channels.
-    shell_channel_class = Type(InProcessChannel)  # type:ignore[arg-type]
-    iopub_channel_class = Type(InProcessChannel)  # type:ignore[arg-type]
-    stdin_channel_class = Type(InProcessChannel)  # type:ignore[arg-type]
-    control_channel_class = Type(InProcessChannel)  # type:ignore[arg-type]
-    hb_channel_class = Type(InProcessHBChannel)  # type:ignore[arg-type]
+    shell_channel_class = Type(InProcessChannel)  # type:ignore[assignment]
+    iopub_channel_class = Type(InProcessChannel)  # type:ignore[assignment]
+    stdin_channel_class = Type(InProcessChannel)  # type:ignore[assignment]
+    control_channel_class = Type(InProcessChannel)  # type:ignore[assignment]
+    hb_channel_class = Type(InProcessHBChannel)  # type:ignore[assignment]
 
     kernel = Instance("ipykernel.inprocess.ipkernel.InProcessKernel", allow_none=True)
 
@@ -57,9 +58,9 @@ def _default_blocking_class(self):
 
         return BlockingInProcessKernelClient
 
-    def get_connection_info(self):
+    def get_connection_info(self, session: bool = False) -> dict[str, int | str | bytes]:
         """Get the connection info for the client."""
-        d = super().get_connection_info()
+        d = super().get_connection_info(session=session)
         d["kernel"] = self.kernel  # type:ignore[assignment]
         return d
 
@@ -72,39 +73,45 @@ def start_channels(self, *args, **kwargs):
     @property
     def shell_channel(self):
         if self._shell_channel is None:
-            self._shell_channel = self.shell_channel_class(self)  # type:ignore[abstract,call-arg]
+            self._shell_channel = self.shell_channel_class(self)
         return self._shell_channel
 
     @property
     def iopub_channel(self):
         if self._iopub_channel is None:
-            self._iopub_channel = self.iopub_channel_class(self)  # type:ignore[abstract,call-arg]
+            self._iopub_channel = self.iopub_channel_class(self)
         return self._iopub_channel
 
     @property
     def stdin_channel(self):
         if self._stdin_channel is None:
-            self._stdin_channel = self.stdin_channel_class(self)  # type:ignore[abstract,call-arg]
+            self._stdin_channel = self.stdin_channel_class(self)
         return self._stdin_channel
 
     @property
     def control_channel(self):
         if self._control_channel is None:
-            self._control_channel = self.control_channel_class(self)  # type:ignore[abstract,call-arg]
+            self._control_channel = self.control_channel_class(self)
         return self._control_channel
 
     @property
     def hb_channel(self):
         if self._hb_channel is None:
-            self._hb_channel = self.hb_channel_class(self)  # type:ignore[abstract,call-arg]
+            self._hb_channel = self.hb_channel_class(self)
         return self._hb_channel
 
     # Methods for sending specific messages
     # -------------------------------------
 
     def execute(
-        self, code, silent=False, store_history=True, user_expressions=None, allow_stdin=None
-    ):
+        self,
+        code: str,
+        silent: bool = False,
+        store_history: bool = True,
+        user_expressions: dict[str, Any] | None = None,
+        allow_stdin: bool | None = None,
+        stop_on_error: bool = True,
+    ) -> str:
         """Execute code on the client."""
         if allow_stdin is None:
             allow_stdin = self.allow_stdin
@@ -117,7 +124,9 @@ def execute(
         )
         msg = self.session.msg("execute_request", content)
         self._dispatch_to_kernel(msg)
-        return msg["header"]["msg_id"]
+        res = msg["header"]["msg_id"]
+        assert isinstance(res, str)
+        return res
 
     def complete(self, code, cursor_pos=None):
         """Get code completion."""

ipykernel/inprocess/ipkernel.py

@@ -51,7 +51,7 @@ class InProcessKernel(IPythonKernel):
     _underlying_iopub_socket = Instance(DummySocket, ())
     iopub_thread: IOPubThread = Instance(IOPubThread)  # type:ignore[assignment]
 
-    shell_stream = Instance(DummySocket, ())  # type:ignore[arg-type]
+    shell_stream = Instance(DummySocket, ())  # type:ignore[assignment]
 
     @default("iopub_thread")
     def _default_iopub_thread(self):
@@ -65,7 +65,7 @@ def _default_iopub_thread(self):
     def _default_iopub_socket(self):
         return self.iopub_thread.background_socket
 
-    stdin_socket = Instance(DummySocket, ())  # type:ignore[assignment]
+    stdin_socket = Instance(DummySocket, ())
 
     def __init__(self, **traits):
         """Initialize the kernel."""
@@ -85,7 +85,7 @@ def start(self):
         if self.shell:
             self.shell.exit_now = False
 
-    def _abort_queues(self):
+    def _abort_queues(self, subshell_id: str | None = ...):
         """The in-process kernel doesn't abort requests."""
 
     def _input_request(self, prompt, ident, parent, password=False):

ipykernel/ipkernel.py

@@ -231,7 +231,7 @@ def dispatch_debugpy(self, msg):
             self.debugger.tcp_client.receive_dap_frame(frame)
 
     @property
-    def banner(self):
+    def banner(self):  # type:ignore[override]
         if self.shell:
             return self.shell.banner
         return None
@@ -455,7 +455,7 @@ async def run_cell(*args, **kwargs):
                     if threading.current_thread() == threading.main_thread()
                     else self._dummy_context_manager
                 )
-                with cm(coro_future):  # type:ignore[operator]
+                with cm(coro_future):
                     res = None
                     try:
                         res = await coro_future

ipykernel/kernelapp.py

@@ -118,9 +118,9 @@ class IPKernelApp(BaseIPythonApplication, InteractiveShellApp, ConnectionFileMix
     """The IPYKernel application class."""
 
     name = "ipython-kernel"
-    aliases = Dict(kernel_aliases)  # type:ignore[assignment]
-    flags = Dict(kernel_flags)  # type:ignore[assignment]
-    classes = [IPythonKernel, ZMQInteractiveShell, ProfileDir, Session]
+    aliases = Dict(kernel_aliases)
+    flags = Dict(kernel_flags)
+    classes = [IPythonKernel, ZMQInteractiveShell, ProfileDir, Session]  # type:ignore[assignment]
     # the kernel class, as an importstring
     kernel_class = Type(
         "ipykernel.ipkernel.IPythonKernel",
@@ -261,7 +261,7 @@ def _bind_socket(self, s, port):
                     raise
         return None
 
-    def write_connection_file(self):
+    def write_connection_file(self, **kwargs: Any) -> None:
         """write connection info to JSON file"""
         cf = self.abs_connection_file
         connection_info = dict(

ipykernel/zmqshell.py

@@ -17,6 +17,7 @@
 import os
 import sys
 import threading
+import typing
 import warnings
 from pathlib import Path
 
@@ -78,12 +79,16 @@ def _hooks(self):
             self._thread_local.hooks = []
         return self._thread_local.hooks
 
-    def publish(
+    # Feb: 2025 IPython has a deprecated, `source` parameter, marked for removal that
+    # triggers typing errors.
+    def publish(  # type:ignore[override]
         self,
         data,
         metadata=None,
+        *,
         transient=None,
         update=False,
+        **kwargs,
     ):
         """Publish a display-data message
 
@@ -125,7 +130,7 @@ def publish(
         for hook in self._hooks:
             msg = hook(msg)
             if msg is None:
-                return  # type:ignore[unreachable]
+                return
 
         self.session.send(
             self.pub_socket,
@@ -153,7 +158,7 @@ def clear_output(self, wait=False):
         for hook in self._hooks:
             msg = hook(msg)
             if msg is None:
-                return  # type:ignore[unreachable]
+                return
 
         self.session.send(
             self.pub_socket,
@@ -482,7 +487,7 @@ def __init__(self, *args, **kwargs):
 
     displayhook_class = Type(ZMQShellDisplayHook)
     display_pub_class = Type(ZMQDisplayPublisher)
-    data_pub_class = Any()  # type:ignore[assignment]
+    data_pub_class = Any()
     kernel = Any()
     parent_header = Any()
 
@@ -520,7 +525,7 @@ def _update_exit_now(self, change):
 
     # Over ZeroMQ, GUI control isn't done with PyOS_InputHook as there is no
     # interactive input being read; we provide event loop support in ipkernel
-    def enable_gui(self, gui):
+    def enable_gui(self, gui: typing.Any = None) -> None:
         """Enable a given guil."""
         from .eventloops import enable_gui as real_enable_gui
 
@@ -590,7 +595,7 @@ def data_pub(self):
                 stacklevel=2,
             )
 
-            self._data_pub = self.data_pub_class(parent=self)  # type:ignore[has-type]
+            self._data_pub = self.data_pub_class(parent=self)
             self._data_pub.session = self.display_pub.session  # type:ignore[attr-defined]
             self._data_pub.pub_socket = self.display_pub.pub_socket  # type:ignore[attr-defined]
         return self._data_pub
@@ -660,14 +665,10 @@ def set_parent(self, parent):
         self.display_pub.set_parent(parent)  # type:ignore[attr-defined]
         if hasattr(self, "_data_pub"):
             self.data_pub.set_parent(parent)
-        try:
-            sys.stdout.set_parent(parent)  # type:ignore[attr-defined]
-        except AttributeError:
-            pass
-        try:
-            sys.stderr.set_parent(parent)  # type:ignore[attr-defined]
-        except AttributeError:
-            pass
+        if hasattr(sys.stdout, "set_parent"):
+            sys.stdout.set_parent(parent)
+        if hasattr(sys.stderr, "set_parent"):
+            sys.stderr.set_parent(parent)
 
     def get_parent(self):
         """Get the parent header."""