Merge pull request #791 from joouha/mypy-any-fixes-1

Correct `Any` type annotations.

Author: davidbrochart

jupyter_client/channels.py

@@ -101,6 +101,7 @@ def _create_socket(self) -> None:
         assert self.context is not None
         self.socket = self.context.socket(zmq.REQ)
         self.socket.linger = 1000
+        assert self.address is not None
         self.socket.connect(self.address)
 
         self.poller.register(self.socket, zmq.POLLIN)
@@ -192,9 +193,7 @@ def call_handlers(self, since_last_heartbeat: float) -> None:
 class ZMQSocketChannel(object):
     """A ZMQ socket in an async API"""
 
-    def __init__(
-        self, socket: zmq.sugar.socket.Socket, session: Session, loop: t.Any = None
-    ) -> None:
+    def __init__(self, socket: zmq.asyncio.Socket, session: Session, loop: t.Any = None) -> None:
         """Create a channel.
 
         Parameters
@@ -208,7 +207,7 @@ def __init__(
         """
         super().__init__()
 
-        self.socket: t.Optional[zmq.sugar.socket.Socket] = socket
+        self.socket: t.Optional[zmq.asyncio.Socket] = socket
         self.session = session
 
     async def _recv(self, **kwargs: t.Any) -> t.Dict[str, t.Any]:

jupyter_client/client.py

@@ -137,19 +137,19 @@ def __del__(self):
     # Channel proxy methods
     # --------------------------------------------------------------------------
 
-    async def _async_get_shell_msg(self, *args: Any, **kwargs: Any) -> t.Dict[str, t.Any]:
+    async def _async_get_shell_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the shell channel"""
         return await self.shell_channel.get_msg(*args, **kwargs)
 
-    async def _async_get_iopub_msg(self, *args: Any, **kwargs: Any) -> t.Dict[str, t.Any]:
+    async def _async_get_iopub_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the iopub channel"""
         return await self.iopub_channel.get_msg(*args, **kwargs)
 
-    async def _async_get_stdin_msg(self, *args: Any, **kwargs: Any) -> t.Dict[str, t.Any]:
+    async def _async_get_stdin_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the stdin channel"""
         return await self.stdin_channel.get_msg(*args, **kwargs)
 
-    async def _async_get_control_msg(self, *args: Any, **kwargs: Any) -> t.Dict[str, t.Any]:
+    async def _async_get_control_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the control channel"""
         return await self.control_channel.get_msg(*args, **kwargs)
 
@@ -270,7 +270,7 @@ def _output_hook_kernel(
         self,
         session: Session,
         socket: zmq.sugar.socket.Socket,
-        parent_header: Any,
+        parent_header: t.Any,
         msg: t.Dict[str, t.Any],
     ) -> None:
         """Output hook when running inside an IPython kernel
@@ -687,7 +687,7 @@ def history(
         raw: bool = True,
         output: bool = False,
         hist_access_type: str = "range",
-        **kwargs: Any,
+        **kwargs: t.Any,
     ) -> str:
         """Get entries from the kernel's history list.
 

jupyter_client/manager.py

@@ -235,7 +235,7 @@ def remove_restart_callback(self, callback: t.Callable, event: str = "restart")
     # create a Client connected to our Kernel
     # --------------------------------------------------------------------------
 
-    def client(self, **kwargs: Any) -> KernelClient:
+    def client(self, **kwargs: t.Any) -> KernelClient:
         """Create a client configured to connect to our kernel"""
         kw = {}
         kw.update(self.get_connection_info(session=True))
@@ -296,7 +296,7 @@ def from_ns(match):
 
         return [pat.sub(from_ns, arg) for arg in cmd]
 
-    async def _async_launch_kernel(self, kernel_cmd: t.List[str], **kw: Any) -> None:
+    async def _async_launch_kernel(self, kernel_cmd: t.List[str], **kw: t.Any) -> None:
         """actually launch the kernel
 
         override in a subclass to launch kernel subprocesses differently
@@ -324,7 +324,9 @@ def _close_control_socket(self) -> None:
         self._control_socket.close()
         self._control_socket = None
 
-    async def _async_pre_start_kernel(self, **kw: Any) -> t.Tuple[t.List[str], t.Dict[str, t.Any]]:
+    async def _async_pre_start_kernel(
+        self, **kw: t.Any
+    ) -> t.Tuple[t.List[str], t.Dict[str, t.Any]]:
         """Prepares a kernel for startup in a separate process.
 
         If random ports (port=0) are being used, this method must be called
@@ -352,7 +354,7 @@ async def _async_pre_start_kernel(self, **kw: Any) -> t.Tuple[t.List[str], t.Dic
 
     pre_start_kernel = run_sync(_async_pre_start_kernel)
 
-    async def _async_post_start_kernel(self, **kw: Any) -> None:
+    async def _async_post_start_kernel(self, **kw: t.Any) -> None:
         """Performs any post startup tasks relative to the kernel.
 
         Parameters
@@ -368,7 +370,7 @@ async def _async_post_start_kernel(self, **kw: Any) -> None:
     post_start_kernel = run_sync(_async_post_start_kernel)
 
     @in_pending_state
-    async def _async_start_kernel(self, **kw: Any) -> None:
+    async def _async_start_kernel(self, **kw: t.Any) -> None:
         """Starts a kernel on this host in a separate process.
 
         If random ports (port=0) are being used, this method must be called
@@ -500,7 +502,7 @@ async def _async_shutdown_kernel(self, now: bool = False, restart: bool = False)
     shutdown_kernel = run_sync(_async_shutdown_kernel)
 
     async def _async_restart_kernel(
-        self, now: bool = False, newports: bool = False, **kw: Any
+        self, now: bool = False, newports: bool = False, **kw: t.Any
     ) -> None:
         """Restarts a kernel with the arguments that were used to launch it.
 
@@ -661,7 +663,7 @@ class AsyncKernelManager(KernelManager):
 
 
 def start_new_kernel(
-    startup_timeout: float = 60, kernel_name: str = "python", **kwargs: Any
+    startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
 ) -> t.Tuple[KernelManager, KernelClient]:
     """Start a new kernel, and return its Manager and Client"""
     km = KernelManager(kernel_name=kernel_name)
@@ -679,7 +681,7 @@ def start_new_kernel(
 
 
 async def start_new_async_kernel(
-    startup_timeout: float = 60, kernel_name: str = "python", **kwargs: Any
+    startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
 ) -> t.Tuple[AsyncKernelManager, KernelClient]:
     """Start a new kernel, and return its Manager and Client"""
     km = AsyncKernelManager(kernel_name=kernel_name)
@@ -697,7 +699,7 @@ async def start_new_async_kernel(
 
 
 @contextmanager
-def run_kernel(**kwargs: Any) -> t.Iterator[KernelClient]:
+def run_kernel(**kwargs: t.Any) -> t.Iterator[KernelClient]:
     """Context manager to create a kernel in a subprocess.
 
     The kernel is shut down when the context exits.

jupyter_client/multikernelmanager.py

@@ -34,7 +34,7 @@ def kernel_method(f: t.Callable) -> t.Callable:
     """decorator for proxying MKM.method(kernel_id) to individual KMs by ID"""
 
     def wrapped(
-        self: Any, kernel_id: str, *args: Any, **kwargs: Any
+        self: t.Any, kernel_id: str, *args: t.Any, **kwargs: t.Any
     ) -> t.Union[t.Callable, t.Awaitable]:
         # get the kernel
         km = self.get_kernel(kernel_id)
@@ -79,7 +79,7 @@ def _kernel_manager_factory_default(self):
     def _create_kernel_manager_factory(self) -> t.Callable:
         kernel_manager_ctor = import_item(self.kernel_manager_class)
 
-        def create_kernel_manager(*args: Any, **kwargs: Any) -> KernelManager:
+        def create_kernel_manager(*args: t.Any, **kwargs: t.Any) -> KernelManager:
             if self.shared_context:
                 if self.context.closed:
                     # recreate context if closed
@@ -142,7 +142,7 @@ def __contains__(self, kernel_id: str) -> bool:
         return kernel_id in self._kernels
 
     def pre_start_kernel(
-        self, kernel_name: t.Optional[str], kwargs: Any
+        self, kernel_name: t.Optional[str], kwargs: t.Any
     ) -> t.Tuple[KernelManager, str, str]:
         # kwargs should be mutable, passing it as a dict argument.
         kernel_id = kwargs.pop("kernel_id", self.new_kernel_id(**kwargs))
@@ -192,7 +192,9 @@ def _using_pending_kernels(self):
         """
         return getattr(self, 'use_pending_kernels', False)
 
-    async def _async_start_kernel(self, kernel_name: t.Optional[str] = None, **kwargs: Any) -> str:
+    async def _async_start_kernel(
+        self, kernel_name: t.Optional[str] = None, **kwargs: t.Any
+    ) -> str:
         """Start a new kernel.
 
         The caller can pick a kernel_id by passing one in as a keyword arg,
@@ -517,7 +519,7 @@ def connect_hb(self, kernel_id: str, identity: t.Optional[bytes] = None) -> sock
         stream : zmq Socket or ZMQStream
         """
 
-    def new_kernel_id(self, **kwargs: Any) -> str:
+    def new_kernel_id(self, **kwargs: t.Any) -> str:
         """
         Returns the id to associate with the kernel for this request. Subclasses may override
         this method to substitute other sources of kernel ids.

jupyter_client/session.py

@@ -216,7 +216,7 @@ class SessionFactory(LoggingConfigurable):
     logname = Unicode("")
 
     @observe("logname")  # type:ignore[misc]
-    def _logname_changed(self, change: Any) -> None:
+    def _logname_changed(self, change: t.Any) -> None:
         self.log = logging.getLogger(change["new"])
 
     # not configurable:
@@ -1077,7 +1077,7 @@ def deserialize(
         # adapt to the current version
         return adapt(message)
 
-    def unserialize(self, *args: Any, **kwargs: Any) -> t.Dict[str, t.Any]:
+    def unserialize(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         warnings.warn(
             "Session.unserialize is deprecated. Use Session.deserialize.",
             DeprecationWarning,

jupyter_client/threaded.py

@@ -8,11 +8,9 @@
 from threading import Event
 from threading import Thread
 from typing import Any
-from typing import Awaitable
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Union
 
 import zmq
 from traitlets import Instance
@@ -30,10 +28,6 @@
 # during garbage collection of threads at exit
 
 
-async def get_msg(msg: Awaitable) -> Union[List[bytes], List[zmq.Message]]:
-    return await msg
-
-
 class ThreadedZMQSocketChannel(object):
     """A ZMQ socket invoking a callback in the ioloop"""
 
@@ -68,6 +62,7 @@ def __init__(
         evt = Event()
 
         def setup_stream():
+            assert self.socket is not None
             self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
             self.stream.on_recv(self._handle_recv)
             evt.set()
@@ -113,13 +108,11 @@ def thread_send():
         assert self.ioloop is not None
         self.ioloop.add_callback(thread_send)
 
-    def _handle_recv(self, future_msg: Awaitable) -> None:
+    def _handle_recv(self, msg_list: List[bytes]) -> None:
         """Callback for stream.on_recv.
 
         Unpacks message, and calls handlers with it.
         """
-        assert self.ioloop is not None
-        msg_list = self.ioloop._asyncio_event_loop.run_until_complete(get_msg(future_msg))
         assert self.session is not None
         ident, smsg = self.session.feed_identities(msg_list)
         msg = self.session.deserialize(smsg)

pyproject.toml

@@ -28,7 +28,7 @@ dependencies = [
     "jupyter_core>=4.9.2",
     "nest-asyncio>=1.5.4",
     "python-dateutil>=2.8.2",
-    "pyzmq>=22.3",
+    "pyzmq>=23.0",
     "tornado>=6.0",
     "traitlets",
 ]