Drop Python 3.8, update Python versions in tests

.github/workflows/main.yml

@@ -39,18 +39,17 @@ jobs:
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
         python-version:
-          - "3.8"
-          - "3.12"
-          - "3.13-dev"
+          - "3.9"
+          - "3.13"
         include:
           - os: windows-latest
-            python-version: "3.11"
+            python-version: "3.12"
           - os: ubuntu-latest
-            python-version: "pypy-3.9"
+            python-version: "pypy-3.10"
           - os: ubuntu-latest
-            python-version: "3.10"
+            python-version: "3.11"
           - os: macos-latest
-            python-version: "3.9"
+            python-version: "3.10"
     steps:
       - uses: actions/checkout@v4
       - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1

.pre-commit-config.yaml

@@ -37,12 +37,12 @@ repos:
         types_or: [yaml, html, json]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: "v1.8.0"
+    rev: "v1.13.0"
     hooks:
       - id: mypy
         files: jupyter_client
         stages: [manual]
-        args: ["--install-types", "--non-interactive"]
+        args: ["--install-types", "--non-interactive", "--python-version=3.9"]
         additional_dependencies:
           ["traitlets>=5.13", "ipykernel>=6.26", "jupyter_core>=5.3.2"]
 

jupyter_client/_version.py

@@ -1,14 +1,14 @@
 """The version information for jupyter client."""
 import re
-from typing import List, Union
+from typing import Union
 
 __version__ = "8.6.3"
 
 # Build up version_info tuple for backwards compatibility
 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:

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,32 +59,32 @@ 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"]
         if msg_type in self.msg_type_map:
             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,15 +127,15 @@ 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)
         return msg
 
     # 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

jupyter_client/client.py

@@ -241,7 +241,7 @@ async def _stdin_hook_default(self, msg: t.Dict[str, t.Any]) -> None:
         prompt = getpass if content.get("password", False) else input
 
         try:
-            raw_data = prompt(content["prompt"])  # type:ignore[operator]
+            raw_data = prompt(content["prompt"])
         except EOFError:
             # turn EOFError into EOF character
             raw_data = "\x04"

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 = (

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,
@@ -65,6 +65,8 @@ def launch_kernel(
     # If this process in running on pythonw, we know that stdin, stdout, and
     # stderr are all invalid.
     redirect_out = sys.executable.endswith("pythonw.exe")
+    _stdout: Any
+    _stderr: Any
     if redirect_out:
         blackhole = open(os.devnull, "w")  # noqa
         _stdout = blackhole if stdout is None else stdout

jupyter_client/localinterfaces.py

@@ -7,8 +7,9 @@
 import re
 import socket
 import subprocess
+from collections.abc import Iterable, Mapping, Sequence
 from subprocess import PIPE, Popen
-from typing import Any, Callable, Iterable, Mapping, Sequence
+from typing import Any, Callable
 from warnings import warn
 
 LOCAL_IPS: list[str] = []