|
1 | 1 | """The IPython kernel implementation""" |
2 | 2 |
|
| 3 | +from __future__ import annotations |
| 4 | + |
3 | 5 | import builtins |
4 | 6 | import gc |
5 | 7 | import getpass |
6 | 8 | import os |
7 | 9 | import sys |
8 | 10 | import threading |
9 | 11 | import typing as t |
| 12 | +from collections.abc import Mapping |
10 | 13 | from dataclasses import dataclass |
11 | 14 |
|
12 | 15 | import comm |
@@ -46,7 +49,7 @@ def _create_comm(*args, **kwargs): |
46 | 49 |
|
47 | 50 | # there can only be one comm manager in a ipykernel process |
48 | 51 | _comm_lock = threading.Lock() |
49 | | -_comm_manager: t.Optional[CommManager] = None |
| 52 | +_comm_manager: CommManager | None = None |
50 | 53 |
|
51 | 54 |
|
52 | 55 | def _get_comm_manager(*args, **kwargs): |
@@ -84,7 +87,7 @@ def _user_module_changed(self, change): |
84 | 87 | if self.shell is not None: |
85 | 88 | self.shell.user_module = change["new"] |
86 | 89 |
|
87 | | - user_ns = Instance(dict, args=None, allow_none=True) |
| 90 | + user_ns = Instance(Mapping, args=None, allow_none=True) |
88 | 91 |
|
89 | 92 | @observe("user_ns") |
90 | 93 | @observe_compat |
@@ -353,7 +356,7 @@ async def do_execute( |
353 | 356 |
|
354 | 357 | self._forward_input(allow_stdin) |
355 | 358 |
|
356 | | - reply_content: t.Dict[str, t.Any] = {} |
| 359 | + reply_content: dict[str, t.Any] = {} |
357 | 360 | if hasattr(shell, "run_cell_async") and hasattr(shell, "should_run_async"): |
358 | 361 | run_cell = shell.run_cell_async |
359 | 362 | should_run_async = shell.should_run_async |
@@ -559,7 +562,7 @@ def do_inspect(self, code, cursor_pos, detail_level=0, omit_sections=()): |
559 | 562 | """Handle code inspection.""" |
560 | 563 | name = token_at_cursor(code, cursor_pos) |
561 | 564 |
|
562 | | - reply_content: t.Dict[str, t.Any] = {"status": "ok"} |
| 565 | + reply_content: dict[str, t.Any] = {"status": "ok"} |
563 | 566 | reply_content["data"] = {} |
564 | 567 | reply_content["metadata"] = {} |
565 | 568 | assert self.shell is not None |
@@ -755,7 +758,7 @@ def init_closure(self: threading.Thread, *args, **kwargs): |
755 | 758 | threading.Thread.run = run_closure # type:ignore[method-assign] |
756 | 759 |
|
757 | 760 | def _clean_thread_parent_frames( |
758 | | - self, phase: t.Literal["start", "stop"], info: t.Dict[str, t.Any] |
| 761 | + self, phase: t.Literal["start", "stop"], info: dict[str, t.Any] |
759 | 762 | ): |
760 | 763 | """Clean parent frames of threads which are no longer running. |
761 | 764 | This is meant to be invoked by garbage collector callback hook. |
|
0 commit comments