fix some linting with ruff

Author: Carreau

enterprise_gateway/base/handlers.py

@@ -5,7 +5,6 @@
 
 
 import json
-from typing import List
 
 import jupyter_server._version
 from jupyter_server.base.handlers import APIHandler
@@ -48,4 +47,4 @@ def prepare(self):
         raise web.HTTPError(404)
 
 
-default_handlers: List[tuple] = [(r"/api", APIVersionHandler), (r"/(.*)", NotFoundHandler)]
+default_handlers: list[tuple] = [(r"/api", APIVersionHandler), (r"/(.*)", NotFoundHandler)]

enterprise_gateway/client/gateway_client.py

@@ -45,7 +45,7 @@ def start_kernel(
         self, kernelspec_name, username=DEFAULT_USERNAME, timeout=REQUEST_TIMEOUT, extra_env=None
     ):
         """Start a kernel."""
-        self.log.info(f"Starting a {kernelspec_name} kernel ....")
+        self.log.info("Starting a %s kernel ....", kernelspec_name)
 
         if extra_env is None:
             extra_env = {}
@@ -65,11 +65,9 @@ def start_kernel(
         if response.status_code == 201:
             json_data = response.json()
             kernel_id = json_data.get("id")
-            self.log.info(f"Started kernel with id {kernel_id}")
+            self.log.info("Started kernel with id %s", kernel_id)
         else:
-            msg = "Error starting kernel : {} response code \n {}".format(
-                response.status_code, response.content
-            )
+            msg = f"Error starting kernel : {response.status_code} response code \n {response.content}"
             raise RuntimeError(msg)
 
         return KernelClient(
@@ -82,7 +80,7 @@ def start_kernel(
 
     def shutdown_kernel(self, kernel):
         """Shut down a kernel."""
-        self.log.info(f"Shutting down kernel : {kernel.kernel_id} ....")
+        self.log.info("Shutting down kernel : %s ....", kernel.kernel_id)
 
         if not kernel:
             return False
@@ -113,7 +111,7 @@ def __init__(
         self.response_reader = Thread(target=self._read_responses)
         self.response_queues = {}
         self.interrupt_thread = None
-        self.log.debug(f"Initializing kernel client ({kernel_id}) to {self.kernel_ws_api_endpoint}")
+        self.log.debug("Initializing kernel client (%s) to %s", kernel_id, self.kernel_ws_api_endpoint)
 
         try:
             self.kernel_socket = websocket.create_connection(
@@ -149,7 +147,7 @@ def shutdown(self):
         url = f"{self.http_api_endpoint}/{self.kernel_id}"
         response = requests.delete(url, timeout=60)
         if response.status_code == 204:
-            self.log.info(f"Kernel {self.kernel_id} shutdown")
+            self.log.info("Kernel %s shutdown", self.kernel_id)
             return True
         else:
             msg = f"Error shutting down kernel {self.kernel_id}: {response.content}"
@@ -209,9 +207,7 @@ def execute(self, code, timeout=REQUEST_TIMEOUT):
                             continue
                     else:
                         self.log.debug(
-                            "Unhandled response for msg_id: {} of msg_type: {}".format(
-                                msg_id, response_message_type
-                            )
+                            f"Unhandled response for msg_id: {msg_id} of msg_type: {response_message_type}"
                         )
 
                 if (
@@ -233,7 +229,7 @@ def interrupt(self):
         url = "{}/{}".format(self.kernel_http_api_endpoint, "interrupt")
         response = requests.post(url, timeout=60)
         if response.status_code == 204:
-            self.log.debug(f"Kernel {self.kernel_id} interrupted")
+            self.log.debug("Kernel %s interrupted", self.kernel_id)
             return True
         else:
             msg = f"Unexpected response interrupting kernel {self.kernel_id}: {response.content}"
@@ -247,7 +243,7 @@ def restart(self, timeout=REQUEST_TIMEOUT):
         url = "{}/{}".format(self.kernel_http_api_endpoint, "restart")
         response = requests.post(url, timeout=60)
         if response.status_code == 200:
-            self.log.debug(f"Kernel {self.kernel_id} restarted")
+            self.log.debug("Kernel %s restarted", self.kernel_id)
             self.kernel_socket = websocket.create_connection(
                 self.kernel_ws_api_endpoint, timeout=timeout, enable_multithread=True
             )
@@ -265,12 +261,10 @@ def get_state(self):
         response = requests.get(url, timeout=60)
         if response.status_code == 200:
             json = response.json()
-            self.log.debug(f"Kernel {self.kernel_id} state: {json}")
+            self.log.debug("Kernel %s state: %s", self.kernel_id, json)
             return json["execution_state"]
         else:
-            msg = "Unexpected response retrieving state for kernel {}: {}".format(
-                self.kernel_id, response.content
-            )
+            msg = f"Unexpected response retrieving state for kernel {self.kernel_id}: {response.content}"
             raise RuntimeError(msg)
 
     def start_interrupt_thread(self, wait_time=DEFAULT_INTERRUPT_WAIT):
@@ -320,7 +314,7 @@ def _get_response(self, msg_id, timeout, post_idle):
 
         msg_queue = self.response_queues.get(msg_id)
         try:
-            self.log.debug(f"Getting response for msg_id: {msg_id} with timeout: {timeout}")
+            self.log.debug("Getting response for msg_id: %s with timeout: %s", msg_id, timeout)
             response = msg_queue.get(timeout=timeout)
             self.log.debug(
                 "Got response for msg_id: {}, msg_type: {}".format(
@@ -366,7 +360,7 @@ def _read_responses(self):
                         i = 1
                         while self.restarting:
                             if i >= 10 and i % 2 == 0:
-                                self.log.debug(f"Still restarting after {i} secs...")
+                                self.log.debug("Still restarting after %s secs...", i)
                             time.sleep(1)
                             i += 1
                         continue
@@ -377,7 +371,7 @@ def _read_responses(self):
 
         except BaseException as be2:
             if not self.shutting_down:
-                self.log.warning(f"Unexpected exception encountered ({be2})")
+                self.log.warning("Unexpected exception encountered (%s)", be2)
 
         self.log.debug("Response reader thread exiting...")
 

enterprise_gateway/enterprisegatewayapp.py

@@ -12,7 +12,7 @@
 import sys
 import time
 import weakref
-from typing import ClassVar, List, Optional
+from typing import ClassVar, Optional
 
 from jupyter_client.kernelspec import KernelSpecManager
 from jupyter_core.application import JupyterApp, base_aliases
@@ -83,7 +83,7 @@ class EnterpriseGatewayApp(EnterpriseGatewayConfigMixin, JupyterApp):
     # Enable some command line shortcuts
     aliases = aliases
 
-    def initialize(self, argv: Optional[List[str]] = None) -> None:
+    def initialize(self, argv: Optional[list[str]] = None) -> None:
         """Initializes the base class, configurable manager instances, the
         Tornado web app, and the tornado HTTP server.
 
@@ -161,7 +161,7 @@ def init_configurables(self) -> None:
 
         self.init_dynamic_configs()
 
-    def _create_request_handlers(self) -> List[tuple]:
+    def _create_request_handlers(self) -> list[tuple]:
         """Create default Jupyter handlers and redefine them off of the
         base_url path. Assumes init_configurables() has already been called.
         """
@@ -215,7 +215,7 @@ def init_webapp(self) -> None:
         handlers = self._create_request_handlers()
 
         # Instantiate the configured authorizer class
-        self.log.info(f"Using authorizer: {self.authorizer_class}")
+        self.log.info("Using authorizer: %s", self.authorizer_class)
         authorizer = self.authorizer_class(parent=self, log=self.log)
 
         self.web_app = web.Application(
@@ -285,10 +285,10 @@ def init_http_server(self) -> None:
                 self.http_server.listen(port, self.ip)
             except OSError as e:
                 if e.errno == errno.EADDRINUSE:
-                    self.log.info("The port %i is already in use, trying another port." % port)
+                    self.log.info("The port %s is already in use, trying another port.", port)
                     continue
                 elif e.errno in (errno.EACCES, getattr(errno, "WSAEACCES", errno.EACCES)):
-                    self.log.warning("Permission to listen on port %i denied" % port)
+                    self.log.warning("Permission to listen on port %s denied", port)
                     continue
                 else:
                     raise
@@ -317,10 +317,8 @@ def start(self) -> None:
             gateway_user = getpass.getuser()
             if gateway_user.lower() not in self.unauthorized_users:
                 self.log.warning(
-                    "Impersonation is enabled and gateway user '{}' is NOT specified in the set of "
-                    "unauthorized users!  Kernels may execute as that user with elevated privileges.".format(
-                        gateway_user
-                    )
+                    f"Impersonation is enabled and gateway user '{gateway_user}' is NOT specified in the set of "
+                    "unauthorized users!  Kernels may execute as that user with elevated privileges."
                 )
 
         self.io_loop = ioloop.IOLoop.current()
@@ -349,7 +347,7 @@ def shutdown(self) -> None:
                     self.kernel_manager.shutdown_kernel(kid, now=True)
                 )
             except Exception as ex:
-                self.log.warning(f"Failed to shut down kernel {kid}: {ex}")
+                self.log.warning("Failed to shut down kernel %s: %s", kid, ex)
         self.log.info("Shut down complete")
 
     def stop(self) -> None:
@@ -382,7 +380,7 @@ def update_dynamic_configurables(self) -> bool:
         for file in self.loaded_config_files:
             mod_time = int(os.path.getmtime(file))
             if mod_time > self._last_config_update:
-                self.log.debug(f"Config file was updated: {file}!")
+                self.log.debug("Config file was updated: %s!", file)
                 self._last_config_update = mod_time
                 updated = True
 
@@ -433,9 +431,7 @@ def init_dynamic_configs(self) -> None:
             self.add_dynamic_configurable("KernelSessionManager", self.kernel_session_manager)
 
             self.log.info(
-                "Dynamic updates have been configured.  Checking every {} seconds.".format(
-                    self.dynamic_config_interval
-                )
+                f"Dynamic updates have been configured.  Checking every {self.dynamic_config_interval} seconds."
             )
 
             self.log.info(
@@ -444,7 +440,7 @@ def init_dynamic_configs(self) -> None:
             )
             for config, options in self.cli_config.items():
                 for option, value in options.items():
-                    self.log.info(f"    '{config}.{option}': '{value}'")
+                    self.log.info("    '%s.%s': '%s'", config, option, value)
 
             if self.dynamic_config_poller is None:
                 self.dynamic_config_poller = ioloop.PeriodicCallback(

enterprise_gateway/mixins.py

@@ -8,9 +8,10 @@
 import os
 import ssl
 import traceback
+from collections.abc import Awaitable
 from distutils.util import strtobool
 from http.client import responses
-from typing import Any, Awaitable, ClassVar, Dict, List, Optional, Set
+from typing import Any, ClassVar, Optional
 
 from tornado import web
 from tornado.log import LogFormatter
@@ -426,7 +427,7 @@ def _update_env_process_whitelist(self, change):
     )
 
     @default("inherited_envs")
-    def _inherited_envs_default(self) -> List[str]:
+    def _inherited_envs_default(self) -> list[str]:
         return os.getenv(self.inherited_envs_env, os.getenv("EG_ENV_PROCESS_WHITELIST", "")).split(
             ","
         )
@@ -439,7 +440,7 @@ def _inherited_envs_default(self) -> List[str]:
     )
 
     @default("kernel_headers")
-    def _kernel_headers_default(self) -> List[str]:
+    def _kernel_headers_default(self) -> list[str]:
         default_headers = os.getenv(self.kernel_headers_env)
         return default_headers.split(",") if default_headers else []
 
@@ -455,7 +456,7 @@ def _kernel_headers_default(self) -> List[str]:
     )
 
     @default("remote_hosts")
-    def _remote_hosts_default(self) -> List[str]:
+    def _remote_hosts_default(self) -> list[str]:
         return os.getenv(self.remote_hosts_env, self.remote_hosts_default_value).split(",")
 
     # load_balancing_algorithm
@@ -477,7 +478,7 @@ def _load_balancing_algorithm_default(self) -> str:
         )
 
     @validate("load_balancing_algorithm")
-    def _validate_load_balancing_algorithm(self, proposal: Dict[str, str]) -> str:
+    def _validate_load_balancing_algorithm(self, proposal: dict[str, str]) -> str:
         value = proposal["value"]
         try:
             if value not in ["round-robin", "least-connection"]:
@@ -588,7 +589,7 @@ def _impersonation_enabled_default(self) -> bool:
     )
 
     @default("unauthorized_users")
-    def _unauthorized_users_default(self) -> Set[str]:
+    def _unauthorized_users_default(self) -> set[str]:
         return os.getenv(self.unauthorized_users_env, self.unauthorized_users_default_value).split(
             ","
         )
@@ -606,7 +607,7 @@ def _unauthorized_users_default(self) -> Set[str]:
     )
 
     @default("authorized_users")
-    def _authorized_users_default(self) -> Set[str]:
+    def _authorized_users_default(self) -> set[str]:
         au_env = os.getenv(self.authorized_users_env)
         return au_env.split(",") if au_env is not None else []
 
@@ -686,7 +687,7 @@ def _dynamic_config_interval_default(self) -> int:
         )
 
     @observe("dynamic_config_interval")
-    def _dynamic_config_interval_changed(self, event: Dict[str, Any]) -> None:
+    def _dynamic_config_interval_changed(self, event: dict[str, Any]) -> None:
         prev_val = event["old"]
         self.dynamic_config_interval = event["new"]
         if self.dynamic_config_interval != prev_val:

enterprise_gateway/services/api/handlers.py

@@ -4,7 +4,6 @@
 # Distributed under the terms of the Modified BSD License.
 
 import os
-from typing import List
 
 from jupyter_server.utils import ensure_async
 from tornado import web
@@ -58,7 +57,7 @@ def get_resource_metadata() -> tuple:
         return "swagger.yaml", "text/x-yaml"
 
 
-default_handlers: List[tuple] = [
+default_handlers: list[tuple] = [
     (f"/api/{SpecJsonHandler.get_resource_metadata()[0]}", SpecJsonHandler),
     (f"/api/{APIYamlHandler.get_resource_metadata()[0]}", APIYamlHandler),
 ]

enterprise_gateway/services/kernels/handlers.py

@@ -86,11 +86,9 @@ async def post(self):
                     else:
                         missing_headers.append(name)
 
-            if len(missing_headers):
+            if missing_headers:
                 self.log.warning(
-                    "The following headers specified in 'kernel-headers' were not found: {}".format(
-                        missing_headers
-                    )
+                    f"The following headers specified in 'kernel-headers' were not found: {missing_headers}"
                 )
 
             # No way to override the call to start_kernel on the kernel manager