Apply safe ruff auto-fixes

krassowski · krassowski · commit c7ec378593e2 · 2025-11-28T11:58:48.000Z
diff --git a/jupyter_server/__init__.py b/jupyter_server/__init__.py
@@ -17,12 +17,12 @@
 from .base.call_context import CallContext
 
 __all__ = [
+    "DEFAULT_EVENTS_SCHEMA_PATH",
+    "DEFAULT_JUPYTER_SERVER_PORT",
     "DEFAULT_STATIC_FILES_PATH",
     "DEFAULT_TEMPLATE_PATH_LIST",
-    "DEFAULT_JUPYTER_SERVER_PORT",
     "JUPYTER_SERVER_EVENTS_URI",
-    "DEFAULT_EVENTS_SCHEMA_PATH",
+    "CallContext",
     "__version__",
     "version_info",
-    "CallContext",
 ]
diff --git a/jupyter_server/auth/decorator.py b/jupyter_server/auth/decorator.py
@@ -82,9 +82,9 @@ async def inner(self, *args, **kwargs):
         method = action
         action = None
         # no-arguments `@authorized` decorator called
-        return cast(FuncT, wrapper(method))
+        return cast("FuncT", wrapper(method))
 
-    return cast(FuncT, wrapper)
+    return cast("FuncT", wrapper)
 
 
 def allow_unauthenticated(method: FuncT) -> FuncT:
@@ -111,7 +111,7 @@ def wrapper(self, *args, **kwargs):
 
     setattr(wrapper, "__allow_unauthenticated", True)
 
-    return cast(FuncT, wrapper)
+    return cast("FuncT", wrapper)
 
 
 def ws_authenticated(method: FuncT) -> FuncT:
@@ -139,4 +139,4 @@ def wrapper(self, *args, **kwargs):
 
     setattr(wrapper, "__allow_unauthenticated", False)
 
-    return cast(FuncT, wrapper)
+    return cast("FuncT", wrapper)
diff --git a/jupyter_server/auth/identity.py b/jupyter_server/auth/identity.py
@@ -252,7 +252,7 @@ async def _get_user(self, handler: web.RequestHandler) -> User | None:
         """Get the user."""
         if getattr(handler, "_jupyter_current_user", None):
             # already authenticated
-            return t.cast(User, handler._jupyter_current_user)  # type:ignore[attr-defined]
+            return t.cast("User", handler._jupyter_current_user)  # type:ignore[attr-defined]
         _token_user: User | None | t.Awaitable[User | None] = self.get_user_token(handler)
         if isinstance(_token_user, t.Awaitable):
             _token_user = await _token_user
@@ -298,7 +298,7 @@ def update_user(
     ) -> User:
         """Update user information and persist the user model."""
         self.check_update(user_data)
-        current_user = t.cast(User, handler.current_user)
+        current_user = t.cast("User", handler.current_user)
         updated_user = self.update_user_model(current_user, user_data)
         self.persist_user_model(handler)
         return updated_user
@@ -585,7 +585,7 @@ def process_login_form(self, handler: web.RequestHandler) -> User | None:
             return self.generate_anonymous_user(handler)
 
         if self.token and self.token == typed_password:
-            return t.cast(User, self.user_for_token(typed_password))  # type:ignore[attr-defined]
+            return t.cast("User", self.user_for_token(typed_password))  # type:ignore[attr-defined]
 
         return user
 
diff --git a/jupyter_server/base/handlers.py b/jupyter_server/base/handlers.py
@@ -75,7 +75,7 @@ def json_sys_info():
 def log() -> Logger:
     """Get the application log."""
     if Application.initialized():
-        return cast(Logger, Application.instance().log)
+        return cast("Logger", Application.instance().log)
     else:
         return app_log
 
@@ -85,7 +85,7 @@ class AuthenticatedHandler(web.RequestHandler):
 
     @property
     def base_url(self) -> str:
-        return cast(str, self.settings.get("base_url", "/"))
+        return cast("str", self.settings.get("base_url", "/"))
 
     @property
     def content_security_policy(self) -> str:
@@ -95,7 +95,7 @@ def content_security_policy(self) -> str:
         """
         if "Content-Security-Policy" in self.settings.get("headers", {}):
             # user-specified, don't override
-            return cast(str, self.settings["headers"]["Content-Security-Policy"])
+            return cast("str", self.settings["headers"]["Content-Security-Policy"])
 
         return "; ".join(
             [
@@ -173,7 +173,7 @@ def get_current_user(self) -> str:
                 DeprecationWarning,
                 stacklevel=2,
             )
-            return cast(str, self._jupyter_current_user)
+            return cast("str", self._jupyter_current_user)
         # haven't called get_user in prepare, raise
         raise RuntimeError(msg)
 
@@ -224,7 +224,7 @@ def login_available(self) -> bool:
         whether the user is already logged in or not.
 
         """
-        return cast(bool, self.identity_provider.login_available)
+        return cast("bool", self.identity_provider.login_available)
 
     @property
     def authorizer(self) -> Authorizer:
@@ -302,34 +302,34 @@ def serverapp(self) -> ServerApp | None:
     @property
     def version_hash(self) -> str:
         """The version hash to use for cache hints for static files"""
-        return cast(str, self.settings.get("version_hash", ""))
+        return cast("str", self.settings.get("version_hash", ""))
 
     @property
     def mathjax_url(self) -> str:
-        url = cast(str, self.settings.get("mathjax_url", ""))
+        url = cast("str", self.settings.get("mathjax_url", ""))
         if not url or url_is_absolute(url):
             return url
         return url_path_join(self.base_url, url)
 
     @property
     def mathjax_config(self) -> str:
-        return cast(str, self.settings.get("mathjax_config", "TeX-AMS-MML_HTMLorMML-full,Safe"))
+        return cast("str", self.settings.get("mathjax_config", "TeX-AMS-MML_HTMLorMML-full,Safe"))
 
     @property
     def default_url(self) -> str:
-        return cast(str, self.settings.get("default_url", ""))
+        return cast("str", self.settings.get("default_url", ""))
 
     @property
     def ws_url(self) -> str:
-        return cast(str, self.settings.get("websocket_url", ""))
+        return cast("str", self.settings.get("websocket_url", ""))
 
     @property
     def contents_js_source(self) -> str:
         self.log.debug(
             "Using contents: %s",
             self.settings.get("contents_js_source", "services/contents"),
         )
-        return cast(str, self.settings.get("contents_js_source", "services/contents"))
+        return cast("str", self.settings.get("contents_js_source", "services/contents"))
 
     # ---------------------------------------------------------------
     # Manager objects
@@ -370,7 +370,7 @@ def event_logger(self) -> EventLogger:
     @property
     def allow_origin(self) -> str:
         """Normal Access-Control-Allow-Origin"""
-        return cast(str, self.settings.get("allow_origin", ""))
+        return cast("str", self.settings.get("allow_origin", ""))
 
     @property
     def allow_origin_pat(self) -> str | None:
@@ -380,7 +380,7 @@ def allow_origin_pat(self) -> str | None:
     @property
     def allow_credentials(self) -> bool:
         """Whether to set Access-Control-Allow-Credentials"""
-        return cast(bool, self.settings.get("allow_credentials", False))
+        return cast("bool", self.settings.get("allow_credentials", False))
 
     def set_default_headers(self) -> None:
         """Add CORS headers, if defined"""
diff --git a/jupyter_server/extension/handler.py b/jupyter_server/extension/handler.py
@@ -26,10 +26,10 @@ def get_template(self, name: str) -> Template:
         """Return the jinja template object for a given name"""
         try:
             env = f"{self.name}_jinja2_env"  # type:ignore[attr-defined]
-            template = cast(Template, self.settings[env].get_template(name))  # type:ignore[attr-defined]
+            template = cast("Template", self.settings[env].get_template(name))  # type:ignore[attr-defined]
             return template
         except TemplateNotFound:
-            return cast(Template, super().get_template(name))  # type:ignore[misc]
+            return cast("Template", super().get_template(name))  # type:ignore[misc]
 
 
 class ExtensionHandlerMixin:
@@ -64,12 +64,12 @@ def serverapp(self) -> ServerApp:
     @property
     def log(self) -> Logger:
         if not hasattr(self, "name"):
-            return cast(Logger, super().log)  # type:ignore[misc]
+            return cast("Logger", super().log)  # type:ignore[misc]
         # Attempt to pull the ExtensionApp's log, otherwise fall back to ServerApp.
         try:
-            return cast(Logger, self.extensionapp.log)
+            return cast("Logger", self.extensionapp.log)
         except AttributeError:
-            return cast(Logger, self.serverapp.log)
+            return cast("Logger", self.serverapp.log)
 
     @property
     def config(self) -> Config:
@@ -81,7 +81,7 @@ def server_config(self) -> Config:
 
     @property
     def base_url(self) -> str:
-        return cast(str, self.settings.get("base_url", "/"))
+        return cast("str", self.settings.get("base_url", "/"))
 
     def render_template(self, name: str, **ns) -> str:
         """Override render template to handle static_paths
@@ -90,20 +90,20 @@ def render_template(self, name: str, **ns) -> str:
         (e.g. default error pages)
         make sure our extension-specific static_url is _not_ used.
         """
-        template = cast(Template, self.get_template(name))  # type:ignore[attr-defined]
+        template = cast("Template", self.get_template(name))  # type:ignore[attr-defined]
         ns.update(self.template_namespace)  # type:ignore[attr-defined]
         if template.environment is self.settings["jinja2_env"]:
             # default template environment, use default static_url
             ns["static_url"] = super().static_url  # type:ignore[misc]
-        return cast(str, template.render(**ns))
+        return cast("str", template.render(**ns))
 
     @property
     def static_url_prefix(self) -> str:
         return self.extensionapp.static_url_prefix
 
     @property
     def static_path(self) -> str:
-        return cast(str, self.settings[f"{self.name}_static_paths"])
+        return cast("str", self.settings[f"{self.name}_static_paths"])
 
     def static_url(self, path: str, include_host: bool | None = None, **kwargs: Any) -> str:
         """Returns a static URL for the given relative static file path.
@@ -151,4 +151,4 @@ def static_url(self, path: str, include_host: bool | None = None, **kwargs: Any)
             "static_url_prefix": self.static_url_prefix,
         }
 
-        return base + cast(str, get_url(settings, path, **kwargs))
+        return base + cast("str", get_url(settings, path, **kwargs))
diff --git a/jupyter_server/gateway/managers.py b/jupyter_server/gateway/managers.py
@@ -674,7 +674,7 @@ def stop(self) -> None:
                     msgs.append(msg["msg_type"])
             if self.channel_name == "iopub" and "shutdown_reply" in msgs:
                 return
-            if len(msgs):
+            if msgs:
                 self.log.warning(
                     f"Stopping channel '{self.channel_name}' with {len(msgs)} unprocessed non-status messages: {msgs}."
                 )
diff --git a/jupyter_server/prometheus/metrics.py b/jupyter_server/prometheus/metrics.py
@@ -72,7 +72,7 @@
 
 __all__ = [
     "HTTP_REQUEST_DURATION_SECONDS",
-    "TERMINAL_CURRENTLY_RUNNING_TOTAL",
     "KERNEL_CURRENTLY_RUNNING_TOTAL",
     "SERVER_INFO",
+    "TERMINAL_CURRENTLY_RUNNING_TOTAL",
 ]
diff --git a/jupyter_server/serverapp.py b/jupyter_server/serverapp.py
@@ -1037,7 +1037,7 @@ def _default_ip(self) -> str:
 
     @validate("ip")
     def _validate_ip(self, proposal: t.Any) -> str:
-        value = t.cast(str, proposal["value"])
+        value = t.cast("str", proposal["value"])
         if value == "*":
             value = ""
         return value
@@ -1539,7 +1539,7 @@ def _deprecated_cookie_config(self, change: t.Any) -> None:
 
     @validate("base_url")
     def _update_base_url(self, proposal: t.Any) -> str:
-        value = t.cast(str, proposal["value"])
+        value = t.cast("str", proposal["value"])
         if not value.startswith("/"):
             value = "/" + value
         if not value.endswith("/"):
@@ -2335,8 +2335,7 @@ def init_resources(self) -> None:
         soft = self.min_open_files_limit
         hard = old_hard
         if soft is not None and old_soft < soft:
-            if hard < soft:
-                hard = soft
+            hard = max(hard, soft)
             self.log.debug(
                 f"Raising open file limit: soft {old_soft}->{soft}; hard {old_hard}->{hard}"
             )
@@ -2870,7 +2869,7 @@ async def cleanup_extensions(self) -> None:
 
     def running_server_info(self, kernel_count: bool = True) -> str:
         """Return the current working directory and the server url information"""
-        info = t.cast(str, self.contents_manager.info_string()) + "\n"
+        info = t.cast("str", self.contents_manager.info_string()) + "\n"
         if kernel_count:
             n_kernels = len(self.kernel_manager.list_kernel_ids())
             kernel_msg = trans.ngettext("%d active kernel", "%d active kernels", n_kernels)
diff --git a/jupyter_server/services/api/handlers.py b/jupyter_server/services/api/handlers.py
@@ -120,7 +120,7 @@ async def get(self):
     @web.authenticated
     async def patch(self):
         """Update user information."""
-        user_data = cast(dict[UpdatableField, str], self.get_json_body())
+        user_data = cast("dict[UpdatableField, str]", self.get_json_body())
         if not user_data:
             raise web.HTTPError(400, "Invalid or missing JSON body")
 
diff --git a/jupyter_server/services/events/handlers.py b/jupyter_server/services/events/handlers.py
@@ -81,12 +81,12 @@ def validate_model(
         if key not in data:
             message = f"Missing `{key}` in the JSON request body."
             raise Exception(message)
-    schema_id = cast(str, data.get("schema_id"))
+    schema_id = cast("str", data.get("schema_id"))
     # The case where a given schema_id isn't found,
     # jupyter_events raises a useful error, so there's no need to
     # handle that case here.
     schema = registry.get(schema_id)
-    version = str(cast(str, data.get("version")))
+    version = str(cast("str", data.get("version")))
     if schema.version != version:
         message = f"Unregistered version: {version!r}≠{schema.version!r} for `{schema_id}`"
         raise Exception(message)
@@ -126,7 +126,7 @@ async def post(self):
         try:
             validate_model(payload, self.event_logger.schemas)
             self.event_logger.emit(
-                schema_id=cast(str, payload.get("schema_id")),
+                schema_id=cast("str", payload.get("schema_id")),
                 data=cast("dict[str, Any]", payload.get("data")),
                 timestamp_override=get_timestamp(payload),
             )
diff --git a/jupyter_server/services/kernels/connection/channels.py b/jupyter_server/services/kernels/connection/channels.py
@@ -288,7 +288,7 @@ async def _register_session(self):
             self.kernel_id in self.multi_kernel_manager
         ):  # only update open sessions if kernel is actively managed
             self._open_sessions[self.session_key] = t.cast(
-                KernelWebsocketHandler, self.websocket_handler
+                "KernelWebsocketHandler", self.websocket_handler
             )
 
     async def prepare(self):
diff --git a/jupyter_server/services/kernels/kernelmanager.py b/jupyter_server/services/kernels/kernelmanager.py
@@ -10,7 +10,7 @@
 
 import asyncio
 import os
-import pathlib  # noqa: TCH003
+import pathlib  # noqa: TC003
 import sys
 import typing as t
 import warnings
@@ -247,7 +247,7 @@ async def _async_start_kernel(  # type:ignore[override]
             self.log.debug(
                 "Kernel args (excluding env): %r", {k: v for k, v in kwargs.items() if k != "env"}
             )
-            env = kwargs.get("env", None)
+            env = kwargs.get("env")
             if env and isinstance(env, dict):  # type:ignore[unreachable]
                 self.log.debug("Kernel argument 'env' passed with: %r", list(env.keys()))  # type:ignore[unreachable]
 
diff --git a/jupyter_server/services/sessions/sessionmanager.py b/jupyter_server/services/sessions/sessionmanager.py
@@ -291,7 +291,7 @@ 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
@@ -345,7 +345,7 @@ async def start_kernel_for_session(
             kernel_name=kernel_name,
             env=kernel_env,
         )
-        return cast(str, kernel_id)
+        return cast("str", kernel_id)
 
     async def save_session(self, session_id, path=None, name=None, type=None, kernel_id=None):
         """Saves the items for the session with the given session_id
diff --git a/jupyter_server/utils.py b/jupyter_server/utils.py
@@ -146,8 +146,7 @@ def to_api_path(os_path: str, root: str = "") -> ApiPath:
     If given, root will be removed from the path.
     root must be a filesystem path already.
     """
-    if os_path.startswith(root):
-        os_path = os_path[len(root) :]
+    os_path = os_path.removeprefix(root)
     parts = os_path.strip(os.path.sep).split(os.path.sep)
     parts = [p for p in parts if p != ""]  # remove duplicate splits
     path = "/".join(parts)
diff --git a/tests/services/contents/test_manager.py b/tests/services/contents/test_manager.py
@@ -188,7 +188,7 @@ def test_invalid_root_dir(jp_file_contents_manager_class, tmp_path):
 def test_get_os_path(jp_file_contents_manager_class, tmp_path):
     fm = jp_file_contents_manager_class(root_dir=str(tmp_path))
     path = fm._get_os_path("/path/to/notebook/test.ipynb")
-    rel_path_list = "/path/to/notebook/test.ipynb".split("/")
+    rel_path_list = ["", "path", "to", "notebook", "test.ipynb"]
     fs_path = os.path.join(fm.root_dir, *rel_path_list)
     assert path == fs_path
 
diff --git a/tests/unix_sockets/test_serverapp_integration.py b/tests/unix_sockets/test_serverapp_integration.py

Original file line number	Diff line number	Diff line change
`@@ -674,7 +674,7 @@ def stop(self) -> None:`
`674`	`674`	`msgs.append(msg["msg_type"])`
`675`	`675`	`if self.channel_name == "iopub" and "shutdown_reply" in msgs:`
`676`	`676`	`return`
`677`		`- if len(msgs):`
	`677`	`+ if msgs:`
`678`	`678`	`self.log.warning(`
`679`	`679`	`f"Stopping channel '{self.channel_name}' with {len(msgs)} unprocessed non-status messages: {msgs}."`
`680`	`680`	`)`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@`
`72`	`72`
`73`	`73`	`__all__ = [`
`74`	`74`	`"HTTP_REQUEST_DURATION_SECONDS",`
`75`		`- "TERMINAL_CURRENTLY_RUNNING_TOTAL",`
`76`	`75`	`"KERNEL_CURRENTLY_RUNNING_TOTAL",`
`77`	`76`	`"SERVER_INFO",`
	`77`	`+ "TERMINAL_CURRENTLY_RUNNING_TOTAL",`
`78`	`78`	`]`