pre-commit

jupyter_server/_version.py

@@ -4,7 +4,6 @@
 """
 
 import re
-from typing import List
 
 # Version string must appear intact for automatic versioning
 __version__ = "2.15.0.dev0"
@@ -13,7 +12,7 @@
 pattern = r"(?P<major>\d+).(?P<minor>\d+).(?P<patch>\d+)(?P<rest>.*)"
 match = re.match(pattern, __version__)
 assert match is not None
-parts: List[object] = [int(match[part]) for part in ["major", "minor", "patch"]]
+parts: list[object] = [int(match[part]) for part in ["major", "minor", "patch"]]
 if match["rest"]:
     parts.append(match["rest"])
 version_info = tuple(parts)

jupyter_server/auth/authorizer.py

@@ -10,7 +10,8 @@
 # Distributed under the terms of the Modified BSD License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Awaitable
+from collections.abc import Awaitable
+from typing import TYPE_CHECKING
 
 from traitlets import Instance
 from traitlets.config import LoggingConfigurable

jupyter_server/base/call_context.py

@@ -3,7 +3,7 @@
 # Distributed under the terms of the Modified BSD License.
 
 from contextvars import Context, ContextVar, copy_context
-from typing import Any, Dict, List
+from typing import Any
 
 
 class CallContext:
@@ -22,7 +22,7 @@ class CallContext:
     # easier management over maintaining a set of ContextVar instances, since the Context is a
     # map of ContextVar instances to their values, and the "name" is no longer a lookup key.
     _NAME_VALUE_MAP = "_name_value_map"
-    _name_value_map: ContextVar[Dict[str, Any]] = ContextVar(_NAME_VALUE_MAP)
+    _name_value_map: ContextVar[dict[str, Any]] = ContextVar(_NAME_VALUE_MAP)
 
     @classmethod
     def get(cls, name: str) -> Any:
@@ -65,7 +65,7 @@ def set(cls, name: str, value: Any) -> None:
         name_value_map[name] = value
 
     @classmethod
-    def context_variable_names(cls) -> List[str]:
+    def context_variable_names(cls) -> list[str]:
         """Returns a list of variable names set for this call context.
 
         Returns
@@ -77,7 +77,7 @@ def context_variable_names(cls) -> List[str]:
         return list(name_value_map.keys())
 
     @classmethod
-    def _get_map(cls) -> Dict[str, Any]:
+    def _get_map(cls) -> dict[str, Any]:
         """Get the map of names to their values from the _NAME_VALUE_MAP context var.
 
         If the map does not exist in the current context, an empty map is created and returned.

jupyter_server/base/handlers.py

@@ -13,9 +13,10 @@
 import re
 import types
 import warnings
+from collections.abc import Awaitable, Coroutine, Sequence
 from http.client import responses
 from logging import Logger
-from typing import TYPE_CHECKING, Any, Awaitable, Coroutine, Sequence, cast
+from typing import TYPE_CHECKING, Any, cast
 from urllib.parse import urlparse
 
 import prometheus_client

jupyter_server/config_manager.py

@@ -14,7 +14,7 @@
 from traitlets.config import LoggingConfigurable
 from traitlets.traitlets import Bool, Unicode
 
-StrDict = t.Dict[str, t.Any]
+StrDict = dict[str, t.Any]
 
 
 def recursive_update(target: StrDict, new: StrDict) -> None:

jupyter_server/files/handlers.py

@@ -6,7 +6,7 @@
 
 import mimetypes
 from base64 import decodebytes
-from typing import Awaitable
+from collections.abc import Awaitable
 
 from jupyter_core.utils import ensure_async
 from tornado import web

jupyter_server/services/api/handlers.py

@@ -4,7 +4,7 @@
 # Distributed under the terms of the Modified BSD License.
 import json
 import os
-from typing import Any, Dict, List
+from typing import Any
 
 from jupyter_core.utils import ensure_async
 from tornado import web
@@ -87,7 +87,7 @@ async def get(self):
         else:
             permissions_to_check = {}
 
-        permissions: Dict[str, List[str]] = {}
+        permissions: dict[str, list[str]] = {}
         user = self.current_user
 
         for resource, actions in permissions_to_check.items():
@@ -106,7 +106,7 @@ async def get(self):
                 if authorized:
                     allowed.append(action)
 
-        identity: Dict[str, Any] = self.identity_provider.identity_model(user)
+        identity: dict[str, Any] = self.identity_provider.identity_model(user)
         model = {
             "identity": identity,
             "permissions": permissions,

jupyter_server/services/config/manager.py

@@ -23,7 +23,7 @@ class ConfigManager(LoggingConfigurable):
 
     def get(self, section_name):
         """Get the config from all config sections."""
-        config: t.Dict[str, t.Any] = {}
+        config: dict[str, t.Any] = {}
         # step through back to front, to ensure front of the list is top priority
         for p in self.read_config_path[::-1]:
             cm = BaseJSONConfigManager(config_dir=p)

jupyter_server/services/contents/handlers.py

@@ -7,7 +7,7 @@
 # Distributed under the terms of the Modified BSD License.
 import json
 from http import HTTPStatus
-from typing import Any, Dict, List
+from typing import Any
 
 try:
     from jupyter_client.jsonutil import json_default
@@ -24,7 +24,7 @@
 AUTH_RESOURCE = "contents"
 
 
-def _validate_keys(expect_defined: bool, model: Dict[str, Any], keys: List[str]):
+def _validate_keys(expect_defined: bool, model: dict[str, Any], keys: list[str]):
     """
     Validate that the keys are defined (i.e. not None) or not (i.e. None)
     """

jupyter_server/services/events/handlers.py

@@ -7,7 +7,7 @@
 
 import json
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Dict, Optional, cast
+from typing import TYPE_CHECKING, Any, Optional, cast
 
 from jupyter_core.utils import ensure_async
 from tornado import web, websocket
@@ -127,7 +127,7 @@ async def post(self):
             validate_model(payload, self.event_logger.schemas)
             self.event_logger.emit(
                 schema_id=cast(str, payload.get("schema_id")),
-                data=cast("Dict[str, Any]", payload.get("data")),
+                data=cast("dict[str, Any]", payload.get("data")),
                 timestamp_override=get_timestamp(payload),
             )
             self.set_status(204)

jupyter_server/services/kernels/connection/abc.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any, List
+from typing import Any
 
 
 class KernelWebsocketConnectionABC(ABC):
@@ -25,5 +25,5 @@ def handle_incoming_message(self, incoming_msg: str) -> None:
         """Broker the incoming websocket message to the appropriate ZMQ channel."""
 
     @abstractmethod
-    def handle_outgoing_message(self, stream: str, outgoing_msg: List[Any]) -> None:
+    def handle_outgoing_message(self, stream: str, outgoing_msg: list[Any]) -> None:
         """Broker outgoing ZMQ messages to the kernel websocket."""

jupyter_server/services/kernels/connection/base.py

@@ -2,7 +2,7 @@
 
 import json
 import struct
-from typing import Any, List
+from typing import Any
 
 from jupyter_client.session import Session
 from tornado.websocket import WebSocketHandler
@@ -89,7 +89,7 @@ def serialize_msg_to_ws_v1(msg_or_list, channel, pack=None):
     else:
         msg_list = msg_or_list
     channel = channel.encode("utf-8")
-    offsets: List[Any] = []
+    offsets: list[Any] = []
     offsets.append(8 * (1 + 1 + len(msg_list) + 1))
     offsets.append(len(channel) + offsets[-1])
     for msg in msg_list:
@@ -173,7 +173,7 @@ def handle_incoming_message(self, incoming_msg: str) -> None:
         """Handle an incoming message."""
         raise NotImplementedError
 
-    def handle_outgoing_message(self, stream: str, outgoing_msg: List[Any]) -> None:
+    def handle_outgoing_message(self, stream: str, outgoing_msg: list[Any]) -> None:
         """Handle an outgoing message."""
         raise NotImplementedError
 

jupyter_server/services/sessions/sessionmanager.py

@@ -5,7 +5,7 @@
 import os
 import pathlib
 import uuid
-from typing import Any, Dict, List, NewType, Optional, Union, cast
+from typing import Any, NewType, Optional, Union, cast
 
 KernelName = NewType("KernelName", str)
 ModelName = NewType("ModelName", str)
@@ -100,7 +100,7 @@ class KernelSessionRecordList:
     it will be appended.
     """
 
-    _records: List[KernelSessionRecord]
+    _records: list[KernelSessionRecord]
 
     def __init__(self, *records: KernelSessionRecord):
         """Initialize a record list."""
@@ -267,7 +267,7 @@ async def create_session(
         type: Optional[str] = None,
         kernel_name: Optional[KernelName] = None,
         kernel_id: Optional[str] = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Creates a session and returns its model
 
         Parameters
@@ -291,11 +291,11 @@ async def create_session(
             session_id, path=path, name=name, type=type, kernel_id=kernel_id
         )
         self._pending_sessions.remove(record)
-        return cast(Dict[str, Any], result)
+        return cast(dict[str, Any], result)
 
     def get_kernel_env(
         self, path: Optional[str], name: Optional[ModelName] = None
-    ) -> Dict[str, str]:
+    ) -> dict[str, str]:
         """Return the environment variables that need to be set in the kernel
 
         Parameters

tests/auth/test_authorizer.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import os
-from typing import Awaitable
+from collections.abc import Awaitable
 
 import pytest
 from jupyter_client.kernelspec import NATIVE_KERNEL_NAME

tests/services/api/test_api.py

@@ -1,5 +1,5 @@
 import json
-from typing import Awaitable, Dict, List
+from collections.abc import Awaitable
 from unittest import mock
 
 import pytest
@@ -31,7 +31,7 @@ async def test_get_status(jp_fetch):
 
 
 class MockUser(User):
-    permissions: Dict[str, List[str]]
+    permissions: dict[str, list[str]]
 
 
 class MockIdentityProvider(IdentityProvider):

tests/services/contents/test_manager.py

@@ -3,7 +3,7 @@
 import sys
 import time
 from itertools import combinations
-from typing import Dict, Optional, Tuple
+from typing import Optional
 from unittest.mock import patch
 
 import pytest
@@ -112,7 +112,7 @@ def add_invalid_cell(notebook):
 
 async def prepare_notebook(
     jp_contents_manager: FileContentsManager, make_invalid: Optional[bool] = False
-) -> Tuple[Dict, str]:
+) -> tuple[dict, str]:
     cm = jp_contents_manager
     model = await ensure_async(cm.new_untitled(type="notebook"))
     name = model["name"]
@@ -983,9 +983,10 @@ async def test_nb_validation(jp_contents_manager):
     # successful methods and ensure that calls to the aliased "validate_nb" are
     # zero.  Note that since patching side-effects the validation error case, we'll
     # skip call-count assertions for that portion of the test.
-    with patch("nbformat.validate") as mock_validate, patch(
-        "jupyter_server.services.contents.manager.validate_nb"
-    ) as mock_validate_nb:
+    with (
+        patch("nbformat.validate") as mock_validate,
+        patch("jupyter_server.services.contents.manager.validate_nb") as mock_validate_nb,
+    ):
         # Valid notebook, save, then get
         model = await ensure_async(cm.save(model, path))
         assert "message" not in model

tests/test_gateway.py

@@ -10,7 +10,7 @@
 from http.cookies import SimpleCookie
 from io import BytesIO
 from queue import Empty
-from typing import Any, Dict, Union
+from typing import Any, Union
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -74,7 +74,7 @@ def generate_kernelspec(name):
 #
 # This is used to simulate inconsistency in list results from the Gateway server
 # due to issues like race conditions, bugs, etc.
-omitted_kernels: Dict[str, bool] = {}
+omitted_kernels: dict[str, bool] = {}
 
 
 def generate_model(name):

tests/test_serverapp.py

@@ -154,8 +154,9 @@ async def test_generate_config(tmp_path, jp_configurable_serverapp):
 
 def test_server_password(tmp_path, jp_configurable_serverapp):
     password = "secret"
-    with patch.dict("os.environ", {"JUPYTER_CONFIG_DIR": str(tmp_path)}), patch.object(
-        getpass, "getpass", return_value=password
+    with (
+        patch.dict("os.environ", {"JUPYTER_CONFIG_DIR": str(tmp_path)}),
+        patch.object(getpass, "getpass", return_value=password),
     ):
         app = JupyterPasswordApp(log_level=logging.ERROR)
         app.initialize([])