Update typing for traitlets 5.11 (#977)

* Update typing for traitlets 5.11

* fix typing

Author: blink1073

jupyter_client/asynchronous/client.py

@@ -48,11 +48,11 @@ def _context_default(self) -> zmq.asyncio.Context:
     wait_for_ready = KernelClient._async_wait_for_ready
 
     # The classes to use for the various channels
-    shell_channel_class = Type(AsyncZMQSocketChannel)
-    iopub_channel_class = Type(AsyncZMQSocketChannel)
-    stdin_channel_class = Type(AsyncZMQSocketChannel)
-    hb_channel_class = Type(HBChannel)
-    control_channel_class = Type(AsyncZMQSocketChannel)
+    shell_channel_class = Type(AsyncZMQSocketChannel)  # type:ignore[arg-type]
+    iopub_channel_class = Type(AsyncZMQSocketChannel)  # type:ignore[arg-type]
+    stdin_channel_class = Type(AsyncZMQSocketChannel)  # type:ignore[arg-type]
+    hb_channel_class = Type(HBChannel)  # type:ignore[arg-type]
+    control_channel_class = Type(AsyncZMQSocketChannel)  # type:ignore[arg-type]
 
     _recv_reply = KernelClient._async_recv_reply
 

jupyter_client/blocking/client.py

@@ -44,11 +44,11 @@ class BlockingKernelClient(KernelClient):
     wait_for_ready = run_sync(KernelClient._async_wait_for_ready)
 
     # The classes to use for the various channels
-    shell_channel_class = Type(ZMQSocketChannel)
-    iopub_channel_class = Type(ZMQSocketChannel)
-    stdin_channel_class = Type(ZMQSocketChannel)
-    hb_channel_class = Type(HBChannel)
-    control_channel_class = Type(ZMQSocketChannel)
+    shell_channel_class = Type(ZMQSocketChannel)  # type:ignore[arg-type]
+    iopub_channel_class = Type(ZMQSocketChannel)  # type:ignore[arg-type]
+    stdin_channel_class = Type(ZMQSocketChannel)  # type:ignore[arg-type]
+    hb_channel_class = Type(HBChannel)  # type:ignore[arg-type]
+    control_channel_class = Type(ZMQSocketChannel)  # type:ignore[arg-type]
 
     _recv_reply = run_sync(KernelClient._async_recv_reply)
 

jupyter_client/client.py

@@ -353,9 +353,9 @@ def shell_channel(self) -> t.Any:
             url = self._make_url("shell")
             self.log.debug("connecting shell channel to %s", url)
             socket = self.connect_shell(identity=self.session.bsession)
-            self._shell_channel = self.shell_channel_class(
+            self._shell_channel = self.shell_channel_class(  # type:ignore[call-arg,abstract]
                 socket, self.session, self.ioloop
-            )  # type:ignore[operator]
+            )
         return self._shell_channel
 
     @property
@@ -365,9 +365,9 @@ def iopub_channel(self) -> t.Any:
             url = self._make_url("iopub")
             self.log.debug("connecting iopub channel to %s", url)
             socket = self.connect_iopub()
-            self._iopub_channel = self.iopub_channel_class(
+            self._iopub_channel = self.iopub_channel_class(  # type:ignore[call-arg,abstract]
                 socket, self.session, self.ioloop
-            )  # type:ignore[operator]
+            )
         return self._iopub_channel
 
     @property
@@ -377,9 +377,9 @@ def stdin_channel(self) -> t.Any:
             url = self._make_url("stdin")
             self.log.debug("connecting stdin channel to %s", url)
             socket = self.connect_stdin(identity=self.session.bsession)
-            self._stdin_channel = self.stdin_channel_class(
+            self._stdin_channel = self.stdin_channel_class(  # type:ignore[call-arg,abstract]
                 socket, self.session, self.ioloop
-            )  # type:ignore[operator]
+            )
         return self._stdin_channel
 
     @property
@@ -388,9 +388,9 @@ def hb_channel(self) -> t.Any:
         if self._hb_channel is None:
             url = self._make_url("hb")
             self.log.debug("connecting heartbeat channel to %s", url)
-            self._hb_channel = self.hb_channel_class(
+            self._hb_channel = self.hb_channel_class(  # type:ignore[call-arg,abstract]
                 self.context, self.session, url
-            )  # type:ignore[operator]
+            )
         return self._hb_channel
 
     @property
@@ -400,9 +400,9 @@ def control_channel(self) -> t.Any:
             url = self._make_url("control")
             self.log.debug("connecting control channel to %s", url)
             socket = self.connect_control(identity=self.session.bsession)
-            self._control_channel = self.control_channel_class(
+            self._control_channel = self.control_channel_class(  # type:ignore[call-arg,abstract]
                 socket, self.session, self.ioloop
-            )  # type:ignore[operator]
+            )
         return self._control_channel
 
     async def _async_is_alive(self) -> bool:

jupyter_client/consoleapp.py

@@ -294,7 +294,7 @@ def init_kernel_manager(self) -> None:
 
         # Create a KernelManager and start a kernel.
         try:
-            self.kernel_manager = self.kernel_manager_class(  # type:ignore[operator]
+            self.kernel_manager = self.kernel_manager_class(
                 ip=self.ip,
                 session=self.session,
                 transport=self.transport,

jupyter_client/kernelspec.py

@@ -238,13 +238,9 @@ def _get_kernel_spec_by_name(self, kernel_name, resource_dir):
                 pass
             else:
                 if resource_dir == RESOURCES:
-                    kspec = self.kernel_spec_class(
-                        resource_dir=resource_dir, **get_kernel_dict()
-                    )  # type:ignore[operator]
+                    kspec = self.kernel_spec_class(resource_dir=resource_dir, **get_kernel_dict())
         if not kspec:
-            kspec = self.kernel_spec_class.from_resource_dir(  # type:ignore[attr-defined]
-                resource_dir
-            )
+            kspec = self.kernel_spec_class.from_resource_dir(resource_dir)
 
         if not KPF.instance(parent=self.parent).is_provisioner_available(kspec):
             raise NoSuchKernel(kernel_name)

jupyter_client/kernelspecapp.py

@@ -5,12 +5,10 @@
 import json
 import os.path
 import sys
-import typing as t
 
 from jupyter_core.application import JupyterApp, base_aliases, base_flags
 from traitlets import Bool, Dict, Instance, List, Unicode
 from traitlets.config.application import Application
-from traitlets.config.loader import Config
 
 from . import __version__
 from .kernelspec import KernelSpecManager
@@ -115,7 +113,7 @@ def _kernel_name_default(self):
         "name": "InstallKernelSpec.kernel_name",
         "prefix": "InstallKernelSpec.prefix",
     }
-    aliases.update(base_aliases)  # type:ignore[arg-type]
+    aliases.update(base_aliases)
 
     flags = {
         "user": (
@@ -185,7 +183,7 @@ def _kernel_spec_manager_default(self):
     flags = {
         "f": ({"RemoveKernelSpec": {"force": True}}, force.help),
     }
-    flags.update(JupyterApp.flags)  # type:ignore[has-type]
+    flags.update(JupyterApp.flags)
 
     def parse_command_line(self, argv):
         """Parse the command line args."""
@@ -275,7 +273,7 @@ def start(self):  # pragma: no cover
                         file=sys.stderr,
                     )
                 self.exit(1)
-            self.exit(e)
+            self.exit(e)  # type:ignore[arg-type]
 
 
 class ListProvisioners(JupyterApp):
@@ -321,10 +319,8 @@ class KernelSpecApp(Application):
         }
     )
 
-    aliases: t.Dict[t.Union[str, t.Tuple[str, ...]], t.Union[str, t.Tuple[str, str]]] = {}
-    flags: t.Dict[
-        t.Union[str, t.Tuple[str, ...]], t.Tuple[t.Union[t.Dict[str, t.Any], Config], str]
-    ] = {}
+    aliases = {}
+    flags = {}
 
     def start(self):
         """Start the application."""

jupyter_client/manager.py

@@ -150,7 +150,7 @@ def _kernel_spec_manager_default(self) -> kernelspec.KernelSpecManager:
         return kernelspec.KernelSpecManager(data_dir=self.data_dir)
 
     @observe("kernel_spec_manager")
-    @observe_compat  # type:ignore[misc]
+    @observe_compat
     def _kernel_spec_manager_changed(self, change: t.Dict[str, Instance]) -> None:
         self._kernel_spec = None
 

jupyter_client/provisioning/local_provisioner.py

@@ -5,7 +5,7 @@
 import os
 import signal
 import sys
-from typing import Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from ..connect import KernelConnectionInfo, LocalPortCache
 from ..launcher import launch_kernel
@@ -150,6 +150,8 @@ async def cleanup(self, restart: bool = False) -> None:
                 self.connection_info['control_port'],
             )
             for port in ports:
+                if TYPE_CHECKING:
+                    assert isinstance(port, int)
                 lpc.return_port(port)
 
     async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:

jupyter_client/threaded.py

@@ -333,11 +333,11 @@ def stop_channels(self) -> None:
         if self.ioloop_thread and self.ioloop_thread.is_alive():
             self.ioloop_thread.stop()
 
-    iopub_channel_class = Type(ThreadedZMQSocketChannel)
-    shell_channel_class = Type(ThreadedZMQSocketChannel)
-    stdin_channel_class = Type(ThreadedZMQSocketChannel)
-    hb_channel_class = Type(HBChannel)
-    control_channel_class = Type(ThreadedZMQSocketChannel)
+    iopub_channel_class = Type(ThreadedZMQSocketChannel)  # type:ignore[arg-type]
+    shell_channel_class = Type(ThreadedZMQSocketChannel)  # type:ignore[arg-type]
+    stdin_channel_class = Type(ThreadedZMQSocketChannel)  # type:ignore[arg-type]
+    hb_channel_class = Type(HBChannel)  # type:ignore[arg-type]
+    control_channel_class = Type(ThreadedZMQSocketChannel)  # type:ignore[arg-type]
 
     def is_alive(self) -> bool:
         """Is the kernel process still running?"""

pyproject.toml

@@ -103,7 +103,7 @@ nowarn = "test -W default {args}"
 
 [tool.hatch.envs.typing]
 features = ["test"]
-dependencies = ["mypy>=1.5.1", "traitlets>=5.10.1"]
+dependencies = ["mypy>=1.5.1", "traitlets>=5.11.2", "jupyter_core>=5.3.2"]
 [tool.hatch.envs.typing.scripts]
 test = "mypy --install-types --non-interactive {args:.}"