Correct Any type annotations.

Author: joouha

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,7 @@ 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 +352,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 +368,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 +500,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 +661,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 +679,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 +697,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,7 @@ 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 +517,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,