diff --git a/ipykernel/inprocess/ipkernel.py b/ipykernel/inprocess/ipkernel.py index 13b17217a..cbbe85278 100644 --- a/ipykernel/inprocess/ipkernel.py +++ b/ipykernel/inprocess/ipkernel.py @@ -104,6 +104,7 @@ def _input_request(self, prompt, ident, parent, password=False): assert self.session is not None msg = self.session.msg("input_request", content, parent) for frontend in self.frontends: + assert frontend is not None if frontend.session.session == parent["header"]["session"]: frontend.stdin_channel.call_handlers(msg) break @@ -138,6 +139,7 @@ def _io_dispatch(self, change): assert self.session is not None ident, msg = self.session.recv(self.iopub_socket.io_thread.socket, copy=False) for frontend in self.frontends: + assert frontend is not None frontend.iopub_channel.call_handlers(msg) # ------ Trait initializers ----------------------------------------------- diff --git a/ipykernel/kernelapp.py b/ipykernel/kernelapp.py index de4682f86..af9629015 100644 --- a/ipykernel/kernelapp.py +++ b/ipykernel/kernelapp.py @@ -443,7 +443,7 @@ def log_connection_info(self): self.log.info(line) # also raw print to the terminal if no parent_handle (`ipython kernel`) # unless log-level is CRITICAL (--quiet) - if not self.parent_handle and int(self.log_level) < logging.CRITICAL: + if not self.parent_handle and int(self.log_level) < logging.CRITICAL: # type:ignore[call-overload] print(_ctrl_c_message, file=sys.__stdout__) for line in lines: print(line, file=sys.__stdout__) @@ -700,7 +700,7 @@ def initialize(self, argv=None): except Exception: # Catch exception when initializing signal fails, eg when running the # kernel on a separate thread - if int(self.log_level) < logging.CRITICAL: + if int(self.log_level) < logging.CRITICAL: # type:ignore[call-overload] self.log.error("Unable to initialize signal:", exc_info=True) self.init_kernel() # shell init steps diff --git a/ipykernel/kernelbase.py b/ipykernel/kernelbase.py index 6d06d4ab5..8f3c0a50a 100644 --- a/ipykernel/kernelbase.py +++ b/ipykernel/kernelbase.py @@ -2,6 +2,7 @@ # Copyright (c) IPython Development Team. # Distributed under the terms of the Modified BSD License. +from __future__ import annotations import asyncio import concurrent.futures @@ -80,7 +81,7 @@ class Kernel(SingletonConfigurable): # attribute to override with a GUI eventloop = Any(None) - processes: t.Dict[str, psutil.Process] = {} + processes: dict[str, psutil.Process] = {} @observe("eventloop") def _update_eventloop(self, change): @@ -93,7 +94,7 @@ def _update_eventloop(self, change): profile_dir = Instance("IPython.core.profiledir.ProfileDir", allow_none=True) shell_stream = Instance(ZMQStream, allow_none=True) - shell_streams = List( + shell_streams: List[t.Any] = List( help="""Deprecated shell_streams alias. Use shell_stream .. versionchanged:: 6.0 @@ -153,10 +154,10 @@ def _default_ident(self): # This should be overridden by wrapper kernels that implement any real # language. - language_info: t.Dict[str, object] = {} + language_info: dict[str, object] = {} # any links that should go in the help menu - help_links = List() + help_links: List[dict[str, str]] = List() # Experimental option to break in non-user code. # The ipykernel source is in the call stack, so the user @@ -180,7 +181,7 @@ def _default_ident(self): # track associations with current request _allow_stdin = Bool(False) - _parents = Dict({"shell": {}, "control": {}}) + _parents: Dict[str, t.Any] = Dict({"shell": {}, "control": {}}) _parent_ident = Dict({"shell": b"", "control": b""}) @property @@ -291,7 +292,7 @@ async def poll_control_queue(self): async def _flush_control_queue(self): """Flush the control queue, wait for processing of any pending messages""" - tracer_future: t.Union[concurrent.futures.Future[object], asyncio.Future[object]] + tracer_future: concurrent.futures.Future[object] | asyncio.Future[object] if self.control_thread: control_loop = self.control_thread.io_loop # concurrent.futures.Futures are threadsafe @@ -945,7 +946,7 @@ async def interrupt_request(self, stream, ident, parent): """Handle an interrupt request.""" if not self.session: return - content: t.Dict[str, t.Any] = {"status": "ok"} + content: dict[str, t.Any] = {"status": "ok"} try: self._send_interrupt_children() except OSError as err: diff --git a/pyproject.toml b/pyproject.toml index 1868b40b3..c06ebb18c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -112,7 +112,7 @@ matrix.qt.features = [ [tool.hatch.envs.typing] features = ["test"] -dependencies = ["mypy>=1.6.0", "traitlets>=5.12.0", "ipython>=8.16.1", "jupyter_client>=8.5"] +dependencies = ["mypy>=1.6.0", "traitlets>=5.13.0", "ipython>=8.16.1", "jupyter_client>=8.5"] [tool.hatch.envs.typing.scripts] test = "mypy --install-types --non-interactive {args}"