Update typing for traitlets 5.13

jupyter_server/extension/application.py

@@ -1,4 +1,6 @@
 """An extension application."""
+from __future__ import annotations
+
 import logging
 import re
 import sys
@@ -158,19 +160,21 @@ class method. This method can be set as a entry_point in
 
     @default("open_browser")
     def _default_open_browser(self):
+        assert self.serverapp is not None
         return self.serverapp.config["ServerApp"].get("open_browser", True)
 
     @property
     def config_file_paths(self):
         """Look on the same path as our parent for config files"""
         # rely on parent serverapp, which should control all config loading
+        assert self.serverapp is not None
         return self.serverapp.config_file_paths
 
     # The extension name used to name the jupyter config
     # file, jupyter_{name}_config.
     # This should also match the jupyter subcommand used to launch
     # this extension from the CLI, e.g. `jupyter {name}`.
-    name: t.Union[str, Unicode] = "ExtensionApp"  # type:ignore[assignment]
+    name: str | Unicode = "ExtensionApp"  # type:ignore[assignment]
 
     @classmethod
     def get_extension_package(cls):
@@ -206,7 +210,7 @@ def _default_url(self):
     ]
 
     # A ServerApp is not defined yet, but will be initialized below.
-    serverapp = Any()
+    serverapp: ServerApp | None = Any()  # type:ignore[assignment]
 
     @default("serverapp")
     def _default_serverapp(self):
@@ -242,6 +246,7 @@ def _default_log_format(self):
     @default("static_url_prefix")
     def _default_static_url_prefix(self):
         static_url = f"static/{self.name}/"
+        assert self.serverapp is not None
         return url_path_join(self.serverapp.base_url, static_url)
 
     static_paths = List(
@@ -264,7 +269,9 @@ def _default_static_url_prefix(self):
 
     settings = Dict(help=_i18n("""Settings that will passed to the server.""")).tag(config=True)
 
-    handlers = List(help=_i18n("""Handlers appended to the server.""")).tag(config=True)
+    handlers: List[tuple[t.Any, ...]] = List(
+        help=_i18n("""Handlers appended to the server.""")
+    ).tag(config=True)  # type:ignore[assignment]
 
     def _config_file_name_default(self):
         """The default config file name."""
@@ -295,6 +302,7 @@ def _prepare_config(self):
     def _prepare_settings(self):
         """Prepare the settings."""
         # Make webapp settings accessible to initialize_settings method
+        assert self.serverapp is not None
         webapp = self.serverapp.web_app
         self.settings.update(**webapp.settings)
 
@@ -314,6 +322,7 @@ def _prepare_settings(self):
 
     def _prepare_handlers(self):
         """Prepare the handlers."""
+        assert self.serverapp is not None
         webapp = self.serverapp.web_app
 
         # Get handlers defined by extension subclass.
@@ -352,7 +361,7 @@ def _prepare_handlers(self):
             )
             new_handlers.append(handler)
 
-        webapp.add_handlers(".*$", new_handlers)
+        webapp.add_handlers(".*$", new_handlers)  # type:ignore[arg-type]
 
     def _prepare_templates(self):
         """Add templates to web app settings if extension has templates."""
@@ -372,7 +381,7 @@ def _jupyter_server_config(self):
         base_config["ServerApp"].update(self.serverapp_config)
         return base_config
 
-    def _link_jupyter_server_extension(self, serverapp):
+    def _link_jupyter_server_extension(self, serverapp: ServerApp) -> None:
         """Link the ExtensionApp to an initialized ServerApp.
 
         The ServerApp is stored as an attribute and config
@@ -436,6 +445,7 @@ def start(self):
         """
         super().start()
         # Start the server.
+        assert self.serverapp is not None
         self.serverapp.start()
 
     def current_activity(self):
@@ -447,6 +457,7 @@ async def stop_extension(self):
 
     def stop(self):
         """Stop the underlying Jupyter server."""
+        assert self.serverapp is not None
         self.serverapp.stop()
         self.serverapp.clear_instance()
 

jupyter_server/extension/manager.py

@@ -1,4 +1,6 @@
 """The extension manager."""
+from __future__ import annotations
+
 import importlib
 
 from tornado.gen import multi

jupyter_server/extension/serverextension.py

@@ -355,7 +355,7 @@ def list_server_extensions(self) -> None:
                     self.log.info(f"      {name} {version} {GREEN_OK}")
                 except Exception as err:
                     exc_info = False
-                    if int(self.log_level) <= logging.DEBUG:
+                    if int(self.log_level) <= logging.DEBUG:  # type:ignore[call-overload]
                         exc_info = True
                     self.log.warning(f"      {RED_X} {err}", exc_info=exc_info)
             # Add a blank line between paths.

jupyter_server/gateway/gateway_client.py

@@ -462,7 +462,7 @@ def _gateway_retry_max_default(self):
     )
     gateway_token_renewer_class_env = "JUPYTER_GATEWAY_TOKEN_RENEWER_CLASS"
     gateway_token_renewer_class = Type(
-        klass=GatewayTokenRenewerBase,  # type:ignore[type-abstract]
+        klass=GatewayTokenRenewerBase,
         config=True,
         help="""The class to use for Gateway token renewal. (JUPYTER_GATEWAY_TOKEN_RENEWER_CLASS env var)""",
     )
@@ -546,7 +546,7 @@ def __init__(self, **kwargs):
         """Initialize a gateway client."""
         super().__init__(**kwargs)
         self._connection_args = {}  # initialized on first use
-        self.gateway_token_renewer = self.gateway_token_renewer_class(parent=self, log=self.log)  # type:ignore[operator]
+        self.gateway_token_renewer = self.gateway_token_renewer_class(parent=self, log=self.log)  # type:ignore[abstract]
 
         # store of cookies with store time
         self._cookies: ty.Dict[str, ty.Tuple[Morsel, datetime]] = {}

jupyter_server/gateway/managers.py

@@ -456,7 +456,7 @@ async def refresh_model(self, model=None):
                 # this kernel manager.  The current kernel manager instance may not have
                 # a parent instance if, say, a server extension is using another application
                 # (e.g., papermill) that uses a KernelManager instance directly.
-                self.parent._kernel_connections[self.kernel_id] = int(model["connections"])
+                self.parent._kernel_connections[self.kernel_id] = int(model["connections"])  # type:ignore[index]
 
         self.kernel = model
         return model

jupyter_server/serverapp.py

@@ -1086,7 +1086,7 @@ def _deprecated_token(self, change: t.Any) -> None:
         self._warn_deprecated_config(change, "IdentityProvider")
 
     @default("token")
-    def _deprecated_token_access(self) -> None:
+    def _deprecated_token_access(self) -> str:
         warnings.warn(
             "ServerApp.token config is deprecated in jupyter-server 2.0. Use IdentityProvider.token",
             DeprecationWarning,
@@ -1884,7 +1884,7 @@ def init_configurables(self) -> None:
         self.gateway_config = GatewayClient.instance(parent=self)
 
         if not issubclass(
-            self.kernel_manager_class,  # type:ignore[arg-type]
+            self.kernel_manager_class,
             AsyncMappingKernelManager,
         ):
             warnings.warn(
@@ -1894,7 +1894,7 @@ def init_configurables(self) -> None:
             )
 
         if not issubclass(
-            self.contents_manager_class,  # type:ignore[arg-type]
+            self.contents_manager_class,
             AsyncContentsManager,
         ):
             warnings.warn(
@@ -1925,8 +1925,8 @@ def init_configurables(self) -> None:
                 "because jupyter-client's version does not allow them (should be >8.3.0)."
             )
 
-        self.kernel_manager = self.kernel_manager_class(**kwargs)  # type:ignore[operator]
-        self.contents_manager = self.contents_manager_class(  # type:ignore[operator]
+        self.kernel_manager = self.kernel_manager_class(**kwargs)
+        self.contents_manager = self.contents_manager_class(
             parent=self,
             log=self.log,
         )
@@ -1968,11 +1968,11 @@ def init_configurables(self) -> None:
                 f"Ignoring deprecated config ServerApp.login_handler_class={self.login_handler_class}."
                 " Superseded by ServerApp.identity_provider_class={self.identity_provider_class}."
             )
-        self.identity_provider = self.identity_provider_class(**identity_provider_kwargs)  # type:ignore[operator]
+        self.identity_provider = self.identity_provider_class(**identity_provider_kwargs)
 
         if self.identity_provider_class is LegacyIdentityProvider:
             # legacy config stored the password in tornado_settings
-            self.tornado_settings["password"] = self.identity_provider.hashed_password
+            self.tornado_settings["password"] = self.identity_provider.hashed_password  # type:ignore[attr-defined]
             self.tornado_settings["token"] = self.identity_provider.token
 
         if self._token_set:
@@ -1989,7 +1989,7 @@ def init_configurables(self) -> None:
                 # that means it has some config that should take higher priority than deprecated ServerApp.token
                 self.log.warning("Ignoring deprecated ServerApp.token config")
 
-        self.authorizer = self.authorizer_class(  # type:ignore[operator]
+        self.authorizer = self.authorizer_class(
             parent=self, log=self.log, identity_provider=self.identity_provider
         )
 

jupyter_server/services/contents/manager.py

@@ -1,10 +1,13 @@
 """A base class for contents managers."""
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
+from __future__ import annotations
+
 import itertools
 import json
 import os
 import re
+import typing as t
 import warnings
 from fnmatch import fnmatch
 
@@ -259,8 +262,8 @@ def run_post_save_hook(self, model, os_path):
                 msg = "fUnexpected error while running post hook save: {e}"
                 raise HTTPError(500, msg) from None
 
-    _pre_save_hooks = List()
-    _post_save_hooks = List()
+    _pre_save_hooks: List[t.Any] = List()
+    _post_save_hooks: List[t.Any] = List()
 
     def register_pre_save_hook(self, hook):
         """Register a pre save hook."""

jupyter_server/services/kernels/connection/channels.py

@@ -1,12 +1,13 @@
 """An implementation of a kernel connection."""
+from __future__ import annotations
+
 import asyncio
 import json
 import time
+import typing as t
 import weakref
 from concurrent.futures import Future
 from textwrap import dedent
-from typing import Dict as Dict_t
-from typing import MutableSet, cast
 
 from jupyter_client import protocol_version as client_protocol_version
 from tornado import gen, web
@@ -95,8 +96,8 @@ def write_message(self):
     # class-level registry of open sessions
     # allows checking for conflict on session-id,
     # which is used as a zmq identity and must be unique.
-    _open_sessions: Dict_t[str, KernelWebsocketHandler] = {}
-    _open_sockets: MutableSet["ZMQChannelsWebsocketConnection"] = weakref.WeakSet()
+    _open_sessions: dict[str, KernelWebsocketHandler] = {}
+    _open_sockets: t.MutableSet[ZMQChannelsWebsocketConnection] = weakref.WeakSet()
 
     _kernel_info_future: Future
     _close_future: Future
@@ -127,7 +128,7 @@ def _default_close_future(self):
     # Queue of (time stamp, byte count)
     # Allows you to specify that the byte count should be lowered
     # by a delta amount at some point in the future.
-    _iopub_window_byte_queue = List([])
+    _iopub_window_byte_queue: List[t.Any] = List([])
 
     @classmethod
     async def close_all(cls):
@@ -285,7 +286,7 @@ async def _register_session(self):
         if (
             self.kernel_id in self.multi_kernel_manager
         ):  # only update open sessions if kernel is actively managed
-            self._open_sessions[self.session_key] = cast(
+            self._open_sessions[self.session_key] = t.cast(
                 KernelWebsocketHandler, self.websocket_handler
             )
 

jupyter_server/services/kernels/kernelmanager.py

@@ -5,6 +5,8 @@
 """
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
+from __future__ import annotations
+
 import asyncio
 import os
 import pathlib
@@ -13,8 +15,6 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 from functools import partial, wraps
-from typing import Dict as DictType
-from typing import Optional
 
 from jupyter_client.ioloop.manager import AsyncIOLoopKernelManager
 from jupyter_client.multikernelmanager import AsyncMultiKernelManager, MultiKernelManager
@@ -64,7 +64,7 @@ def _default_kernel_manager_class(self):
 
     _kernel_connections = Dict()
 
-    _kernel_ports: DictType[str, t.List[int]] = Dict()  # type: ignore[assignment]
+    _kernel_ports: dict[str, list[int]] = Dict()  # type: ignore[assignment]
 
     _culler_callback = None
 
@@ -206,7 +206,7 @@ async def _remove_kernel_when_ready(self, kernel_id, kernel_awaitable):
     # TODO DEC 2022: Revise the type-ignore once the signatures have been changed upstream
     # https://github.com/jupyter/jupyter_client/pull/905
     async def _async_start_kernel(  # type:ignore[override]
-        self, *, kernel_id: Optional[str] = None, path: Optional[ApiPath] = None, **kwargs: str
+        self, *, kernel_id: str | None = None, path: ApiPath | None = None, **kwargs: str
     ) -> str:
         """Start a kernel for a session and return its kernel_id.
 
@@ -797,12 +797,12 @@ class ServerKernelManager(AsyncIOLoopKernelManager):
     # schema to register with this kernel manager's eventlogger.
     # This trait should not be overridden.
     @property
-    def core_event_schema_paths(self) -> t.List[pathlib.Path]:
+    def core_event_schema_paths(self) -> list[pathlib.Path]:
         return [DEFAULT_EVENTS_SCHEMA_PATH / "kernel_actions" / "v1.yaml"]
 
     # This trait is intended for subclasses to override and define
     # custom event schemas.
-    extra_event_schema_paths = List(
+    extra_event_schema_paths: List[str] = List(  # type:ignore[assignment]
         default_value=[],
         help="""
         A list of pathlib.Path objects pointing at to register with

pyproject.toml

@@ -104,7 +104,7 @@ nowarn = "test -W default {args}"
 
 [tool.hatch.envs.typing]
 features = ["test"]
-dependencies = [ "mypy~=1.6", "traitlets>=5.11.2", "jupyter_core>=5.3.2", "jupyter_client>=8.5"]
+dependencies = [ "mypy~=1.6", "traitlets>=5.13", "jupyter_core>=5.5", "jupyter_client>=8.5"]
 [tool.hatch.envs.typing.scripts]
 test = "mypy --install-types --non-interactive {args:.}"
 

tests/extension/mockextensions/app.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import os
 
 from jupyter_events import EventLogger
@@ -44,7 +46,7 @@ def get(self):
 
 class MockExtensionApp(ExtensionAppJinjaMixin, ExtensionApp):
     name = "mockextension"
-    template_paths = List().tag(config=True)
+    template_paths: List[str] = List().tag(config=True)  # type:ignore[assignment]
     static_paths = [STATIC_PATH]  # type:ignore[assignment]
     mock_trait = Unicode("mock trait", config=True)
     loaded = False
@@ -59,7 +61,7 @@ def initialize_settings(self):
         # Only add this event if it hasn't already been added.
         # Log the error if it fails, but don't crash the app.
         try:
-            elogger: EventLogger = self.serverapp.event_logger
+            elogger: EventLogger = self.serverapp.event_logger  # type:ignore[union-attr, assignment]
             elogger.register_event_schema(EVENT_SCHEMA)
         except SchemaRegistryException as err:
             self.log.error(err)

tests/services/kernels/test_connection.py

@@ -11,9 +11,9 @@
 from jupyter_server.services.kernels.websocket import KernelWebsocketHandler
 
 
-async def test_websocket_connection(jp_serverapp):
-    app: ServerApp = jp_serverapp
-    kernel_id = await app.kernel_manager.start_kernel()
+async def test_websocket_connection(jp_serverapp: ServerApp) -> None:
+    app = jp_serverapp
+    kernel_id = await app.kernel_manager.start_kernel()  # type:ignore[has-type]
     kernel = app.kernel_manager.get_kernel(kernel_id)
     request = HTTPRequest("foo", "GET")
     request.connection = MagicMock()