strict all the way

Author: Carreau

ipykernel/comm/comm.py

@@ -9,7 +9,7 @@
 
 import comm.base_comm
 import traitlets.config
-from traitlets import Bool, Bytes, Instance, Unicode, default
+from traitlets import Bool, Instance, Unicode, default
 
 from ipykernel.jsonutil import json_clean
 from ipykernel.kernelbase import Kernel
@@ -50,18 +50,18 @@ 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]
-    comm_id = Unicode()
-    primary = Bool(True, help="Am I the primary or secondary Comm?")
+    comm_id = Unicode()  # type: ignore[assignment]
+    primary = Bool(True, help="Am I the primary or secondary Comm?")  # type: ignore[assignment]
 
-    target_name = Unicode("comm")
-    target_module = Unicode(
+    target_name = Unicode("comm")  # type: ignore[assignment]
+    target_module = Unicode(  # type: ignore[assignment]
         None,
         allow_none=True,
         help="""requirejs module from
         which to load comm target.""",
     )
 
-    topic = Bytes()
+    topic: bytes
 
     @default("kernel")
     def _default_kernel(self):

ipykernel/comm/manager.py

@@ -8,6 +8,7 @@
 import comm.base_comm
 import traitlets
 import traitlets.config
+from typing import Dict, Any
 
 from .comm import Comm
 
@@ -18,8 +19,8 @@ class CommManager(comm.base_comm.CommManager, traitlets.config.LoggingConfigurab
     """A comm manager."""
 
     kernel = traitlets.Instance("ipykernel.kernelbase.Kernel")
-    comms = traitlets.Dict()
-    targets = traitlets.Dict()
+    comms: Dict[Any, Any]
+    targets: Dict[str, Any]
 
     def __init__(self, **kwargs):
         """Initialize the manager."""
@@ -33,7 +34,7 @@ def comm_open(self, stream, ident, msg):
         # but we should let the base class create the comm with comm.create_comm in a major release
         content = msg["content"]
         comm_id = content["comm_id"]
-        target_name = content["target_name"]
+        target_name: str = content["target_name"]
         f = self.targets.get(target_name, None)
         comm = Comm(
             comm_id=comm_id,

ipykernel/debugger.py

@@ -21,7 +21,7 @@
 
 try:
     # This import is required to have the next ones working...
-    from debugpy.server import api  # noqa: F401
+    from debugpy.server import api  # type: ignore[import-untyped]# noqa: F401
 
     from _pydevd_bundle import (  # type:ignore[import-not-found]
         pydevd_frame_utils,
@@ -119,9 +119,10 @@ def __init__(self, event_callback, log):
         self.tcp_buffer = ""
         self._reset_tcp_pos()
         self.event_callback = event_callback
-        self.message_send_stream, self.message_receive_stream = create_memory_object_stream[dict](
-            max_buffer_size=inf
-        )
+        (
+            self.message_send_stream,
+            self.message_receive_stream,
+        ) = create_memory_object_stream[dict[t.Any, t.Any]](max_buffer_size=inf)
         self.log = log
 
     def _reset_tcp_pos(self):
@@ -344,9 +345,10 @@ def __init__(
         self.is_started = False
         self.event_callback = event_callback
         self.just_my_code = just_my_code
-        self.stopped_send_stream, self.stopped_receive_stream = create_memory_object_stream[dict](
-            max_buffer_size=inf
-        )
+        (
+            self.stopped_send_stream,
+            self.stopped_receive_stream,
+        ) = create_memory_object_stream[dict[t.Any, t.Any]](max_buffer_size=inf)
 
         self.started_debug_handlers = {}
         for msg_type in Debugger.started_debug_msg_types:

ipykernel/eventloops.py

@@ -279,7 +279,7 @@ def _schedule_exit(delay):
     else:
         import asyncio
 
-        import nest_asyncio
+        import nest_asyncio  # type: ignore [import-untyped]
 
         nest_asyncio.apply()
 

ipykernel/inprocess/ipkernel.py

@@ -82,7 +82,9 @@ async def execute_request(self, stream, ident, parent):
         with self._redirected_io():
             await super().execute_request(stream, ident, parent)
 
-    async def start(self, *, task_status: TaskStatus = TASK_STATUS_IGNORED) -> None:
+    async def start(
+        self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
+    ) -> None:
         """Override registration of dispatchers for streams."""
         if self.shell:
             self.shell.exit_now = False

ipykernel/inprocess/manager.py

@@ -46,7 +46,7 @@ def _default_session(self):
     # --------------------------------------------------------------------------
 
     async def start_kernel(  # type: ignore[explicit-override, override]
-        self, *, task_status: TaskStatus = TASK_STATUS_IGNORED, **kwds: Any
+        self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED, **kwds: Any
     ) -> None:
         """Start the kernel."""
         from ipykernel.inprocess.ipkernel import InProcessKernel
@@ -65,7 +65,7 @@ async def restart_kernel(  # type: ignore[explicit-override, override]
         now: bool = False,
         newports: bool = False,
         *,
-        task_status: TaskStatus = TASK_STATUS_IGNORED,
+        task_status: TaskStatus[None] = TASK_STATUS_IGNORED,
         **kw: Any,
     ) -> None:
         """Restart the kernel."""

ipykernel/inprocess/socket.py

@@ -9,6 +9,7 @@
 import zmq.asyncio
 from anyio import create_memory_object_stream
 from traitlets import HasTraits, Instance
+from typing import Any
 
 # -----------------------------------------------------------------------------
 # Dummy socket class
@@ -32,12 +33,12 @@ def __init__(self, is_shell, *args, **kwargs):
         self.is_shell = is_shell
         self.on_recv = None
         if is_shell:
-            self.in_send_stream, self.in_receive_stream = create_memory_object_stream[dict](
-                max_buffer_size=inf
-            )
-            self.out_send_stream, self.out_receive_stream = create_memory_object_stream[dict](
-                max_buffer_size=inf
-            )
+            self.in_send_stream, self.in_receive_stream = create_memory_object_stream[
+                dict[Any, Any]
+            ](max_buffer_size=inf)
+            self.out_send_stream, self.out_receive_stream = create_memory_object_stream[
+                dict[Any, Any]
+            ](max_buffer_size=inf)
 
     def put(self, msg):
         self.in_send_stream.send_nowait(msg)

ipykernel/ipkernel.py

@@ -246,7 +246,9 @@ async def poll_stopped_queue(self):
         while True:
             await self.debugger.handle_stopped_event()
 
-    async def start(self, *, task_status: TaskStatus = TASK_STATUS_IGNORED) -> None:
+    async def start(
+        self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
+    ) -> None:
         """Start the kernel."""
         if self.shell:
             self.shell.exit_now = False

ipykernel/kernelbase.py

@@ -539,7 +539,9 @@ def pre_handler_hook(self):
     def post_handler_hook(self):
         """Hook to execute after calling message handler"""
 
-    async def start(self, *, task_status: TaskStatus = TASK_STATUS_IGNORED) -> None:
+    async def start(
+        self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
+    ) -> None:
         """Process messages on shell and control channels"""
         async with create_task_group() as tg:
             self.control_stop = threading.Event()

ipykernel/pickleutil.py

@@ -76,7 +76,7 @@ def use_dill():
     adds support for object methods and closures to serialization.
     """
     # import dill causes most of the magic
-    import dill
+    import dill  # type: ignore[import-untyped]
 
     # dill doesn't work with cPickle,
     # tell the two relevant modules to use plain pickle
@@ -100,7 +100,7 @@ def use_cloudpickle():
 
     adds support for object methods and closures to serialization.
     """
-    import cloudpickle
+    import cloudpickle  # type: ignore[import-untyped]
 
     global pickle  # noqa: PLW0603
     pickle = cloudpickle

ipykernel/trio_runner.py

@@ -50,7 +50,7 @@ async def trio_main():
             async with trio.open_nursery() as nursery:
                 # TODO This hack prevents the nursery from cancelling all child
                 # tasks when an uncaught exception occurs, but it's ugly.
-                nursery._add_exc = log_nursery_exc
+                nursery._add_exc = log_nursery_exc  # type: ignore [method-assign]
                 builtins.GLOBAL_NURSERY = nursery  # type:ignore[attr-defined]
                 await trio.sleep_forever()
 

ipykernel/zmqshell.py

@@ -636,12 +636,12 @@ def set_parent(self, parent):
             self.data_pub.set_parent(parent)
         try:
             stdout = sys.stdout
-            stdout.set_parent(parent)
+            stdout.set_parent(parent)  # type: ignore[attr-defined]
         except AttributeError:
             pass
         try:
             stderr = sys.stderr
-            stderr.set_parent(parent)
+            stderr.set_parent(parent)  # type: ignore[attr-defined]
         except AttributeError:
             pass