fix typing

Author: Carreau

jupyter_server/gateway/managers.py

@@ -318,7 +318,7 @@ async def get_kernel_spec_resource(self, kernel_name, path):
 class GatewaySessionManager(SessionManager):
     kernel_manager = Instance("jupyter_server.gateway.managers.GatewayMappingKernelManager")
 
-    async def kernel_culled(self, kernel_id: str) -> bool:
+    async def kernel_culled(self, kernel_id: str) -> bool:  # typing: ignore
         """Checks if the kernel is still considered alive and returns true if it's not found."""
         km: Optional[GatewayKernelManager] = None
         try:

jupyter_server/services/kernels/kernelmanager.py

@@ -7,6 +7,7 @@
 # Distributed under the terms of the Modified BSD License.
 import asyncio
 import os
+import typing as t
 import warnings
 from collections import defaultdict
 from datetime import datetime, timedelta
@@ -58,7 +59,7 @@ def _default_kernel_manager_class(self):
 
     _kernel_connections = Dict()
 
-    _kernel_ports: DictType[str, str] = Dict()  # type: ignore
+    _kernel_ports: DictType[str, t.List[int]] = Dict()  # type: ignore
 
     _culler_callback = None
 
@@ -196,24 +197,30 @@ async def _remove_kernel_when_ready(self, kernel_id, kernel_awaitable):
         self._kernel_connections.pop(kernel_id, None)
         self._kernel_ports.pop(kernel_id, None)
 
+    # note: Kernel_name is not used, but we have to take it as a fist
+    # parameter as this is the first argument taken by the superclass, and the
+    # Liskov Substitution Principle (LSP) states that objects of a superclass
+    # should be replaceable with objects of its subclasses without breaking
+    # the application.
     async def _async_start_kernel(
-        self, kernel_id: Optional[str] = None, path: Optional[ApiPath] = None, **kwargs: str
+        self, kernel_name: Optional[str] = None, path: Optional[ApiPath] = None, **kwargs: str
     ) -> str:
         """Start a kernel for a session and return its kernel_id.
 
         Parameters
         ----------
+        kernel_name : str
+            The name identifying which kernel spec to launch. This is ignored if
+            an existing kernel is returned, but it may be checked in the future.
         kernel_id : uuid (str)
             The uuid to associate the new kernel with. If this
             is not None, this kernel will be persistent whenever it is
             requested.
         path : API path
             The API path (unicode, '/' delimited) for the cwd.
             Will be transformed to an OS path relative to root_dir.
-        kernel_name : str
-            The name identifying which kernel spec to launch. This is ignored if
-            an existing kernel is returned, but it may be checked in the future.
         """
+        kernel_id = kwargs.pop("kernel_id")
         if kernel_id is None or kernel_id not in self:
             if path is not None:
                 kwargs["cwd"] = self.cwd_for_path(path, env=kwargs.get("env", {}))
@@ -301,6 +308,8 @@ def _get_changed_ports(self, kernel_id):
         """
         # Get current ports and return comparison with ports captured at startup.
         km = self.get_kernel(kernel_id)
+        assert isinstance(km.ports, list)
+        assert isinstance(self._kernel_ports[kernel_id], list)
         if km.ports != self._kernel_ports[kernel_id]:
             return km.ports
         return None

jupyter_server/services/sessions/sessionmanager.py

@@ -449,7 +449,7 @@ async def update_session(self, session_id, **kwargs):
         query = "UPDATE session SET %s WHERE session_id=?" % (", ".join(sets))
         self.cursor.execute(query, list(kwargs.values()) + [session_id])
 
-    def kernel_culled(self, kernel_id: str) -> bool:
+    async def kernel_culled(self, kernel_id: str) -> bool:
         """Checks if the kernel is still considered alive and returns true if its not found."""
         return kernel_id not in self.kernel_manager