From 5a5c105cbfc3c2f59f753324636b110ad8dd9b7e Mon Sep 17 00:00:00 2001
From: David Brochart <david.brochart@gmail.com>
Date: Wed, 13 Nov 2024 09:27:33 +0100
Subject: [PATCH] Fix types

---
 jupyter_client/_version.py                    |  4 +-
 jupyter_client/adapter.py                     | 64 +++++++++----------
 jupyter_client/connect.py                     |  6 +-
 jupyter_client/launcher.py                    |  6 +-
 jupyter_client/multikernelmanager.py          |  4 +-
 jupyter_client/provisioning/factory.py        | 10 +--
 .../provisioning/local_provisioner.py         | 14 ++--
 .../provisioning/provisioner_base.py          | 16 ++---
 jupyter_client/ssh/forward.py                 |  6 +-
 jupyter_client/threaded.py                    | 10 +--
 tests/test_provisioning.py                    |  8 +--
 tests/utils.py                                |  3 +-
 12 files changed, 72 insertions(+), 79 deletions(-)

diff --git a/jupyter_client/_version.py b/jupyter_client/_version.py
index 8c309c15..3101ad82 100644
--- a/jupyter_client/_version.py
+++ b/jupyter_client/_version.py
@@ -1,6 +1,6 @@
 """The version information for jupyter client."""
 import re
-from typing import List, Union
+from typing import Union
 
 __version__ = "8.6.3"
 
@@ -8,7 +8,7 @@
 pattern = r"(?P<major>\d+).(?P<minor>\d+).(?P<patch>\d+)(?P<rest>.*)"
 match = re.match(pattern, __version__)
 if match:
-    parts: List[Union[int, str]] = [int(match[part]) for part in ["major", "minor", "patch"]]
+    parts: list[Union[int, str]] = [int(match[part]) for part in ["major", "minor", "patch"]]
     if match["rest"]:
         parts.append(match["rest"])
 else:
diff --git a/jupyter_client/adapter.py b/jupyter_client/adapter.py
index 24201f2c..677251de 100644
--- a/jupyter_client/adapter.py
+++ b/jupyter_client/adapter.py
@@ -3,12 +3,12 @@
 # Distributed under the terms of the Modified BSD License.
 import json
 import re
-from typing import Any, Dict, List, Tuple
+from typing import Any
 
 from ._version import protocol_version_info
 
 
-def code_to_line(code: str, cursor_pos: int) -> Tuple[str, int]:
+def code_to_line(code: str, cursor_pos: int) -> tuple[str, int]:
     """Turn a multiline code block and cursor position into a single line
     and new cursor position.
 
@@ -59,17 +59,17 @@ class Adapter:
     Override message_type(msg) methods to create adapters.
     """
 
-    msg_type_map: Dict[str, str] = {}
+    msg_type_map: dict[str, str] = {}
 
-    def update_header(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def update_header(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Update the header."""
         return msg
 
-    def update_metadata(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def update_metadata(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Update the metadata."""
         return msg
 
-    def update_msg_type(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def update_msg_type(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Update the message type."""
         header = msg["header"]
         msg_type = header["msg_type"]
@@ -77,14 +77,14 @@ def update_msg_type(self, msg: Dict[str, Any]) -> Dict[str, Any]:
             msg["msg_type"] = header["msg_type"] = self.msg_type_map[msg_type]
         return msg
 
-    def handle_reply_status_error(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def handle_reply_status_error(self, msg: dict[str, Any]) -> dict[str, Any]:
         """This will be called *instead of* the regular handler
 
         on any reply with status != ok
         """
         return msg
 
-    def __call__(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def __call__(self, msg: dict[str, Any]) -> dict[str, Any]:
         msg = self.update_header(msg)
         msg = self.update_metadata(msg)
         msg = self.update_msg_type(msg)
@@ -100,7 +100,7 @@ def __call__(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         return handler(msg)
 
 
-def _version_str_to_list(version: str) -> List[int]:
+def _version_str_to_list(version: str) -> list[int]:
     """convert a version string to a list of ints
 
     non-int segments are excluded
@@ -127,7 +127,7 @@ class V5toV4(Adapter):
         "inspect_reply": "object_info_reply",
     }
 
-    def update_header(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def update_header(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Update the header."""
         msg["header"].pop("version", None)
         msg["parent_header"].pop("version", None)
@@ -135,7 +135,7 @@ def update_header(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
     # shell channel
 
-    def kernel_info_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def kernel_info_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a kernel info reply."""
         v4c = {}
         content = msg["content"]
@@ -152,20 +152,20 @@ def kernel_info_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         msg["content"] = v4c
         return msg
 
-    def execute_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def execute_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an execute request."""
         content = msg["content"]
         content.setdefault("user_variables", [])
         return msg
 
-    def execute_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def execute_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an execute reply."""
         content = msg["content"]
         content.setdefault("user_variables", {})
         # TODO: handle payloads
         return msg
 
-    def complete_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def complete_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a complete request."""
         content = msg["content"]
         code = content["code"]
@@ -179,7 +179,7 @@ def complete_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         new_content["cursor_pos"] = cursor_pos
         return msg
 
-    def complete_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def complete_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a complete reply."""
         content = msg["content"]
         cursor_start = content.pop("cursor_start")
@@ -189,7 +189,7 @@ def complete_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         content.pop("metadata", None)
         return msg
 
-    def object_info_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def object_info_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an object info request."""
         content = msg["content"]
         code = content["code"]
@@ -201,20 +201,20 @@ def object_info_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         new_content["detail_level"] = content["detail_level"]
         return msg
 
-    def object_info_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def object_info_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """inspect_reply can't be easily backward compatible"""
         msg["content"] = {"found": False, "oname": "unknown"}
         return msg
 
     # iopub channel
 
-    def stream(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def stream(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a stream message."""
         content = msg["content"]
         content["data"] = content.pop("text")
         return msg
 
-    def display_data(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def display_data(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a display data message."""
         content = msg["content"]
         content.setdefault("source", "display")
@@ -229,7 +229,7 @@ def display_data(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
     # stdin channel
 
-    def input_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def input_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an input request."""
         msg["content"].pop("password", None)
         return msg
@@ -243,7 +243,7 @@ class V4toV5(Adapter):
     # invert message renames above
     msg_type_map = {v: k for k, v in V5toV4.msg_type_map.items()}
 
-    def update_header(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def update_header(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Update the header."""
         msg["header"]["version"] = self.version
         if msg["parent_header"]:
@@ -252,7 +252,7 @@ def update_header(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
     # shell channel
 
-    def kernel_info_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def kernel_info_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a kernel info reply."""
         content = msg["content"]
         for key in ("protocol_version", "ipython_version"):
@@ -275,7 +275,7 @@ def kernel_info_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         content["banner"] = ""
         return msg
 
-    def execute_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def execute_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an execute request."""
         content = msg["content"]
         user_variables = content.pop("user_variables", [])
@@ -284,7 +284,7 @@ def execute_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
             user_expressions[v] = v
         return msg
 
-    def execute_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def execute_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an execute reply."""
         content = msg["content"]
         user_expressions = content.setdefault("user_expressions", {})
@@ -301,7 +301,7 @@ def execute_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
         return msg
 
-    def complete_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def complete_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a complete request."""
         old_content = msg["content"]
 
@@ -310,7 +310,7 @@ def complete_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         new_content["cursor_pos"] = old_content["cursor_pos"]
         return msg
 
-    def complete_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def complete_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a complete reply."""
         # complete_reply needs more context than we have to get cursor_start and end.
         # use special end=null to indicate current cursor position and negative offset
@@ -328,7 +328,7 @@ def complete_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         new_content["metadata"] = {}
         return msg
 
-    def inspect_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def inspect_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an inspect request."""
         content = msg["content"]
         name = content["oname"]
@@ -339,7 +339,7 @@ def inspect_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
         new_content["detail_level"] = content["detail_level"]
         return msg
 
-    def inspect_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def inspect_reply(self, msg: dict[str, Any]) -> dict[str, Any]:
         """inspect_reply can't be easily backward compatible"""
         content = msg["content"]
         new_content = msg["content"] = {"status": "ok"}
@@ -363,13 +363,13 @@ def inspect_reply(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
     # iopub channel
 
-    def stream(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def stream(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle a stream message."""
         content = msg["content"]
         content["text"] = content.pop("data")
         return msg
 
-    def display_data(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def display_data(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle display data."""
         content = msg["content"]
         content.pop("source", None)
@@ -384,13 +384,13 @@ def display_data(self, msg: Dict[str, Any]) -> Dict[str, Any]:
 
     # stdin channel
 
-    def input_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:
+    def input_request(self, msg: dict[str, Any]) -> dict[str, Any]:
         """Handle an input request."""
         msg["content"].setdefault("password", False)
         return msg
 
 
-def adapt(msg: Dict[str, Any], to_version: int = protocol_version_info[0]) -> Dict[str, Any]:
+def adapt(msg: dict[str, Any], to_version: int = protocol_version_info[0]) -> dict[str, Any]:
     """Adapt a single message to a target version
 
     Parameters
diff --git a/jupyter_client/connect.py b/jupyter_client/connect.py
index a634be3d..f5d54550 100644
--- a/jupyter_client/connect.py
+++ b/jupyter_client/connect.py
@@ -16,7 +16,7 @@
 import tempfile
 import warnings
 from getpass import getpass
-from typing import TYPE_CHECKING, Any, Dict, Union, cast
+from typing import TYPE_CHECKING, Any, Union, cast
 
 import zmq
 from jupyter_core.paths import jupyter_data_dir, jupyter_runtime_dir, secure_write
@@ -32,7 +32,7 @@
     from .session import Session
 
 # Define custom type for kernel connection info
-KernelConnectionInfo = Dict[str, Union[int, str, bytes]]
+KernelConnectionInfo = dict[str, Union[int, str, bytes]]
 
 
 def write_connection_file(
@@ -275,7 +275,7 @@ def tunnel_to_kernel(
         with open(connection_info) as f:
             connection_info = json.loads(f.read())
 
-    cf = cast(Dict[str, Any], connection_info)
+    cf = cast(dict[str, Any], connection_info)
 
     lports = tunnel.select_random_ports(5)
     rports = (
diff --git a/jupyter_client/launcher.py b/jupyter_client/launcher.py
index f0d07ad1..1171adcd 100644
--- a/jupyter_client/launcher.py
+++ b/jupyter_client/launcher.py
@@ -5,17 +5,17 @@
 import sys
 import warnings
 from subprocess import PIPE, Popen
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 from traitlets.log import get_logger
 
 
 def launch_kernel(
-    cmd: List[str],
+    cmd: list[str],
     stdin: Optional[int] = None,
     stdout: Optional[int] = None,
     stderr: Optional[int] = None,
-    env: Optional[Dict[str, str]] = None,
+    env: Optional[dict[str, str]] = None,
     independent: bool = False,
     cwd: Optional[str] = None,
     **kw: Any,
diff --git a/jupyter_client/multikernelmanager.py b/jupyter_client/multikernelmanager.py
index d14a3f84..44f20407 100644
--- a/jupyter_client/multikernelmanager.py
+++ b/jupyter_client/multikernelmanager.py
@@ -262,9 +262,7 @@ async def _async_start_kernel(self, *, kernel_name: str | None = None, **kwargs:
         km, kernel_name, kernel_id = self.pre_start_kernel(kernel_name, kwargs)
         if not isinstance(km, KernelManager):
             self.log.warning(  # type:ignore[unreachable]
-                "Kernel manager class ({km_class}) is not an instance of 'KernelManager'!".format(
-                    km_class=self.kernel_manager_class.__class__
-                )
+                f"Kernel manager class ({self.kernel_manager_class.__class__}) is not an instance of 'KernelManager'!"
             )
         kwargs["kernel_id"] = kernel_id  # Make kernel_id available to manager and provisioner
 
diff --git a/jupyter_client/provisioning/factory.py b/jupyter_client/provisioning/factory.py
index bad7c15c..53f21074 100644
--- a/jupyter_client/provisioning/factory.py
+++ b/jupyter_client/provisioning/factory.py
@@ -4,7 +4,7 @@
 import glob
 import sys
 from os import getenv, path
-from typing import Any, Dict, List
+from typing import Any
 
 # See compatibility note on `group` keyword in https://docs.python.org/3/library/importlib.metadata.html#entry-points
 if sys.version_info < (3, 10):  # pragma: no cover
@@ -33,7 +33,7 @@ class KernelProvisionerFactory(SingletonConfigurable):
     """
 
     GROUP_NAME = "jupyter_client.kernel_provisioners"
-    provisioners: Dict[str, EntryPoint] = {}
+    provisioners: dict[str, EntryPoint] = {}
 
     default_provisioner_name_env = "JUPYTER_DEFAULT_PROVISIONER_NAME"
     default_provisioner_name = Unicode(
@@ -122,7 +122,7 @@ def _check_availability(self, provisioner_name: str) -> bool:
                 is_available = False
         return is_available
 
-    def _get_provisioner_config(self, kernel_spec: Any) -> Dict[str, Any]:
+    def _get_provisioner_config(self, kernel_spec: Any) -> dict[str, Any]:
         """
         Return the kernel_provisioner stanza from the kernel_spec.
 
@@ -151,7 +151,7 @@ def _get_provisioner_config(self, kernel_spec: Any) -> Dict[str, Any]:
             return env_provisioner  # Return what we found (plus config stanza if necessary)
         return {"provisioner_name": self.default_provisioner_name, "config": {}}
 
-    def get_provisioner_entries(self) -> Dict[str, str]:
+    def get_provisioner_entries(self) -> dict[str, str]:
         """
         Returns a dictionary of provisioner entries.
 
@@ -164,7 +164,7 @@ def get_provisioner_entries(self) -> Dict[str, str]:
         return entries
 
     @staticmethod
-    def _get_all_provisioners() -> List[EntryPoint]:
+    def _get_all_provisioners() -> list[EntryPoint]:
         """Wrapper around entry_points (to fetch the set of provisioners) - primarily to facilitate testing."""
         return entry_points(group=KernelProvisionerFactory.GROUP_NAME)
 
diff --git a/jupyter_client/provisioning/local_provisioner.py b/jupyter_client/provisioning/local_provisioner.py
index 42d8d32d..3cb57fd0 100644
--- a/jupyter_client/provisioning/local_provisioner.py
+++ b/jupyter_client/provisioning/local_provisioner.py
@@ -5,7 +5,7 @@
 import os
 import signal
 import sys
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 from ..connect import KernelConnectionInfo, LocalPortCache
 from ..launcher import launch_kernel
@@ -154,7 +154,7 @@ async def cleanup(self, restart: bool = False) -> None:
                     assert isinstance(port, int)
                 lpc.return_port(port)
 
-    async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
+    async def pre_launch(self, **kwargs: Any) -> dict[str, Any]:
         """Perform any steps in preparation for kernel process launch.
 
         This includes applying additional substitutions to the kernel launch command and env.
@@ -204,7 +204,7 @@ async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
 
         return await super().pre_launch(cmd=kernel_cmd, **kwargs)
 
-    async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnectionInfo:
+    async def launch_kernel(self, cmd: list[str], **kwargs: Any) -> KernelConnectionInfo:
         """Launch a kernel with a command."""
         scrubbed_kwargs = LocalProvisioner._scrub_kwargs(kwargs)
         self.process = launch_kernel(cmd, **scrubbed_kwargs)
@@ -220,21 +220,21 @@ async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnection
         return self.connection_info
 
     @staticmethod
-    def _scrub_kwargs(kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    def _scrub_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
         """Remove any keyword arguments that Popen does not tolerate."""
-        keywords_to_scrub: List[str] = ["extra_arguments", "kernel_id"]
+        keywords_to_scrub: list[str] = ["extra_arguments", "kernel_id"]
         scrubbed_kwargs = kwargs.copy()
         for kw in keywords_to_scrub:
             scrubbed_kwargs.pop(kw, None)
         return scrubbed_kwargs
 
-    async def get_provisioner_info(self) -> Dict:
+    async def get_provisioner_info(self) -> dict:
         """Captures the base information necessary for persistence relative to this instance."""
         provisioner_info = await super().get_provisioner_info()
         provisioner_info.update({"pid": self.pid, "pgid": self.pgid, "ip": self.ip})
         return provisioner_info
 
-    async def load_provisioner_info(self, provisioner_info: Dict) -> None:
+    async def load_provisioner_info(self, provisioner_info: dict) -> None:
         """Loads the base information necessary for persistence relative to this instance."""
         await super().load_provisioner_info(provisioner_info)
         self.pid = provisioner_info["pid"]
diff --git a/jupyter_client/provisioning/provisioner_base.py b/jupyter_client/provisioning/provisioner_base.py
index eff89432..e5e33f67 100644
--- a/jupyter_client/provisioning/provisioner_base.py
+++ b/jupyter_client/provisioning/provisioner_base.py
@@ -3,7 +3,7 @@
 # Distributed under the terms of the Modified BSD License.
 import os
 from abc import ABC, ABCMeta, abstractmethod
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Optional, Union
 
 from traitlets.config import Instance, LoggingConfigurable, Unicode
 
@@ -103,7 +103,7 @@ async def terminate(self, restart: bool = False) -> None:
         pass
 
     @abstractmethod
-    async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnectionInfo:
+    async def launch_kernel(self, cmd: list[str], **kwargs: Any) -> KernelConnectionInfo:
         """
         Launch the kernel process and return its connection information.
 
@@ -136,7 +136,7 @@ async def shutdown_requested(self, restart: bool = False) -> None:
         """
         pass
 
-    async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
+    async def pre_launch(self, **kwargs: Any) -> dict[str, Any]:
         """
         Perform any steps in preparation for kernel process launch.
 
@@ -170,7 +170,7 @@ async def post_launch(self, **kwargs: Any) -> None:
         """
         pass
 
-    async def get_provisioner_info(self) -> Dict[str, Any]:
+    async def get_provisioner_info(self) -> dict[str, Any]:
         """
         Captures the base information necessary for persistence relative to this instance.
 
@@ -180,12 +180,12 @@ async def get_provisioner_info(self) -> Dict[str, Any]:
 
         NOTE: The superclass method must always be called first to ensure proper serialization.
         """
-        provisioner_info: Dict[str, Any] = {}
+        provisioner_info: dict[str, Any] = {}
         provisioner_info["kernel_id"] = self.kernel_id
         provisioner_info["connection_info"] = self.connection_info
         return provisioner_info
 
-    async def load_provisioner_info(self, provisioner_info: Dict) -> None:
+    async def load_provisioner_info(self, provisioner_info: dict) -> None:
         """
         Loads the base information necessary for persistence relative to this instance.
 
@@ -219,7 +219,7 @@ def get_stable_start_time(self, recommended: float = 10.0) -> float:
         """
         return recommended
 
-    def _finalize_env(self, env: Dict[str, str]) -> None:
+    def _finalize_env(self, env: dict[str, str]) -> None:
         """
         Ensures env is appropriate prior to launch.
 
@@ -233,7 +233,7 @@ def _finalize_env(self, env: Dict[str, str]) -> None:
             # If set, it can bork all the things.
             env.pop("PYTHONEXECUTABLE", None)
 
-    def __apply_env_substitutions(self, substitution_values: Dict[str, str]) -> Dict[str, str]:
+    def __apply_env_substitutions(self, substitution_values: dict[str, str]) -> dict[str, str]:
         """
         Walks entries in the kernelspec's env stanza and applies substitutions from current env.
 
diff --git a/jupyter_client/ssh/forward.py b/jupyter_client/ssh/forward.py
index e2f28d21..dc9e5099 100644
--- a/jupyter_client/ssh/forward.py
+++ b/jupyter_client/ssh/forward.py
@@ -62,11 +62,7 @@ def handle(self):
             return
 
         logger.debug(
-            "Connected!  Tunnel open {!r} -> {!r} -> {!r}".format(
-                self.request.getpeername(),
-                chan.getpeername(),
-                (self.chain_host, self.chain_port),
-            )
+            f"Connected!  Tunnel open {self.request.getpeername()!r} -> {chan.getpeername()!r} -> {(self.chain_host, self.chain_port)!r}"
         )
         while True:
             r, w, x = select.select([self.request, chan], [], [])
diff --git a/jupyter_client/threaded.py b/jupyter_client/threaded.py
index 48e0ae77..b0bf5faf 100644
--- a/jupyter_client/threaded.py
+++ b/jupyter_client/threaded.py
@@ -7,7 +7,7 @@
 from concurrent.futures import Future
 from functools import partial
 from threading import Thread
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 import zmq
 from tornado.ioloop import IOLoop
@@ -118,7 +118,7 @@ def close_stream() -> None:
                 pass
             self.socket = None
 
-    def send(self, msg: Dict[str, Any]) -> None:
+    def send(self, msg: dict[str, Any]) -> None:
         """Queue a message to be sent from the IOLoop's thread.
 
         Parameters
@@ -136,7 +136,7 @@ def thread_send() -> None:
         assert self.ioloop is not None
         self.ioloop.add_callback(thread_send)
 
-    def _handle_recv(self, msg_list: List) -> None:
+    def _handle_recv(self, msg_list: list) -> None:
         """Callback for stream.on_recv.
 
         Unpacks message, and calls handlers with it.
@@ -150,7 +150,7 @@ def _handle_recv(self, msg_list: List) -> None:
             self._inspect(msg)  # type:ignore[unreachable]
         self.call_handlers(msg)
 
-    def call_handlers(self, msg: Dict[str, Any]) -> None:
+    def call_handlers(self, msg: dict[str, Any]) -> None:
         """This method is called in the ioloop thread when a message arrives.
 
         Subclasses should override this method to handle incoming messages.
@@ -322,7 +322,7 @@ def start_channels(
 
         super().start_channels(shell, iopub, stdin, hb, control)
 
-    def _check_kernel_info_reply(self, msg: Dict[str, Any]) -> None:
+    def _check_kernel_info_reply(self, msg: dict[str, Any]) -> None:
         """This is run in the ioloop thread when the kernel info reply is received"""
         if msg["msg_type"] == "kernel_info_reply":
             self._handle_kernel_info_reply(msg)
diff --git a/tests/test_provisioning.py b/tests/test_provisioning.py
index f9c847d0..02d4979b 100644
--- a/tests/test_provisioning.py
+++ b/tests/test_provisioning.py
@@ -7,7 +7,7 @@
 import signal
 import sys
 from subprocess import PIPE
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 import pytest
 from jupyter_core import paths
@@ -93,7 +93,7 @@ async def terminate(self, restart: bool = False) -> None:
         if self.process:
             self.process.terminate()
 
-    async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
+    async def pre_launch(self, **kwargs: Any) -> dict[str, Any]:
         km = self.parent
         if km:
             # save kwargs for use in restart
@@ -112,7 +112,7 @@ async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
             return await super().pre_launch(cmd=kernel_cmd, **kwargs)
         return {}
 
-    async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnectionInfo:
+    async def launch_kernel(self, cmd: list[str], **kwargs: Any) -> KernelConnectionInfo:
         scrubbed_kwargs = kwargs
         self.process = launch_kernel(cmd, **scrubbed_kwargs)
         pgid = None
@@ -200,7 +200,7 @@ def akm(request, all_provisioners):
 }
 
 
-def mock_get_all_provisioners() -> List[EntryPoint]:
+def mock_get_all_provisioners() -> list[EntryPoint]:
     result = []
     for name, epstr in initial_provisioner_map.items():
         result.append(EntryPoint(name, epstr, KernelProvisionerFactory.GROUP_NAME))
diff --git a/tests/utils.py b/tests/utils.py
index 063c5392..de43089a 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -4,7 +4,6 @@
 import json
 import os
 import sys
-from typing import Dict
 
 import pytest
 
@@ -41,7 +40,7 @@ def install_kernel(kernels_dir, argv=None, name="test", display_name=None):
 
 
 class RecordCallMixin:
-    method_calls: Dict[str, int]
+    method_calls: dict[str, int]
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)