Fix ruff validation issues

enterprise_gateway/enterprisegatewayapp.py

@@ -12,7 +12,7 @@
 import sys
 import time
 import weakref
-from typing import List, Optional
+from typing import ClassVar, List, Optional
 
 from jupyter_client.kernelspec import KernelSpecManager
 from jupyter_core.application import JupyterApp, base_aliases
@@ -78,7 +78,7 @@ class EnterpriseGatewayApp(EnterpriseGatewayConfigMixin, JupyterApp):
     """
 
     # Also include when generating help options
-    classes = [
+    classes: ClassVar = [
         KernelSpecCache,
         FileKernelSessionManager,
         WebhookKernelSessionManager,
@@ -369,7 +369,7 @@ def _signal_stop(self, sig, frame) -> None:
         self.io_loop.add_callback_from_signal(self.io_loop.stop)
 
     _last_config_update = int(time.time())
-    _dynamic_configurables = {}
+    _dynamic_configurables: ClassVar = {}
 
     def update_dynamic_configurables(self) -> bool:
         """

enterprise_gateway/mixins.py

@@ -9,7 +9,7 @@
 import traceback
 from distutils.util import strtobool
 from http.client import responses
-from typing import Any, Awaitable, Dict, List, Optional, Set
+from typing import Any, Awaitable, ClassVar, Dict, List, Optional, Set
 
 from tornado import web
 from tornado.log import LogFormatter
@@ -36,7 +36,7 @@ class CORSMixin:
     Mixes CORS headers into tornado.web.RequestHandlers.
     """
 
-    SETTINGS_TO_HEADERS = {
+    SETTINGS_TO_HEADERS: ClassVar = {
         "eg_allow_credentials": "Access-Control-Allow-Credentials",
         "eg_allow_headers": "Access-Control-Allow-Headers",
         "eg_allow_methods": "Access-Control-Allow-Methods",

enterprise_gateway/services/kernels/remotemanager.py

@@ -10,7 +10,7 @@
 import signal
 import time
 import uuid
-from typing import Any
+from typing import Any, ClassVar
 
 from jupyter_client.ioloop.manager import AsyncIOLoopKernelManager
 from jupyter_client.kernelspec import KernelSpec
@@ -136,7 +136,7 @@ class TrackPendingRequests:
     """
 
     _pending_requests_all = 0
-    _pending_requests_user = {}
+    _pending_requests_user: ClassVar = {}
 
     def increment(self, username: str) -> None:
         """Increment the requests for a username."""
@@ -658,7 +658,7 @@ async def signal_kernel(self, signum: int) -> None:
                     if alt_sigint:
                         try:
                             sig_value = getattr(signal, alt_sigint)
-                            if type(sig_value) is int:  # Python 2
+                            if isinstance(sig_value, int):  # Python 2
                                 self.sigint_value = sig_value
                             else:  # Python 3
                                 self.sigint_value = sig_value.value

enterprise_gateway/services/kernelspecs/kernelspec_cache.py

@@ -4,7 +4,7 @@
 
 
 import os
-from typing import Dict, Optional, Union
+from typing import ClassVar, Dict, Optional, Union
 
 from jupyter_client.kernelspec import KernelSpec
 from jupyter_server.utils import ensure_async
@@ -105,9 +105,7 @@ def get_item(self, kernel_name: str) -> Optional[KernelSpec]:
                     pass
             if not kernelspec:
                 self.cache_misses += 1
-                self.log.debug(
-                    f"Cache miss ({self.cache_misses}) for kernelspec: {kernel_name}"
-                )
+                self.log.debug(f"Cache miss ({self.cache_misses}) for kernelspec: {kernel_name}")
         return kernelspec
 
     def get_all_items(self) -> Dict[str, CacheItemType]:
@@ -145,9 +143,7 @@ def put_item(self, kernel_name: str, cache_item: Union[KernelSpec, CacheItemType
             observed_dir = os.path.dirname(resource_dir)
             if observed_dir not in self.observed_dirs:
                 # New directory to watch, schedule it...
-                self.log.debug(
-                    f"KernelSpecCache: observing directory: {observed_dir}"
-                )
+                self.log.debug(f"KernelSpecCache: observing directory: {observed_dir}")
                 self.observed_dirs.add(observed_dir)
                 self.observer.schedule(KernelSpecChangeHandler(self), observed_dir, recursive=True)
 
@@ -182,9 +178,7 @@ def _initialize(self):
             for kernel_dir in self.kernel_spec_manager.kernel_dirs:
                 if kernel_dir not in self.observed_dirs:
                     if os.path.exists(kernel_dir):
-                        self.log.info(
-                            f"KernelSpecCache: observing directory: {kernel_dir}"
-                        )
+                        self.log.info(f"KernelSpecCache: observing directory: {kernel_dir}")
                         self.observed_dirs.add(kernel_dir)
                         self.observer.schedule(
                             KernelSpecChangeHandler(self), kernel_dir, recursive=True
@@ -217,7 +211,7 @@ class KernelSpecChangeHandler(FileSystemEventHandler):
     # Events related to these files trigger the management of the KernelSpec cache.  Should we find
     # other files qualify as indicators of a kernel specification's state (like perhaps detached parameter
     # files in the future) should be added to this list - at which time it should become configurable.
-    watched_files = ["kernel.json"]
+    watched_files: ClassVar = ["kernel.json"]
 
     def __init__(self, kernel_spec_cache: KernelSpecCache, **kwargs):
         """Initialize the handler."""

enterprise_gateway/services/processproxies/conductor.py

@@ -13,7 +13,7 @@
 import subprocess
 import time
 from random import randint
-from typing import Any
+from typing import Any, ClassVar
 
 from jupyter_client import localinterfaces
 from jupyter_server.utils import url_unescape
@@ -32,8 +32,8 @@ class ConductorClusterProcessProxy(RemoteProcessProxy):
     Kernel lifecycle management for Conductor clusters.
     """
 
-    initial_states = {"SUBMITTED", "WAITING", "RUNNING"}
-    final_states = {"FINISHED", "KILLED", "RECLAIMED"}  # Don't include FAILED state
+    initial_states: ClassVar = {"SUBMITTED", "WAITING", "RUNNING"}
+    final_states: ClassVar = {"FINISHED", "KILLED", "RECLAIMED"}  # Don't include FAILED state
 
     def __init__(self, kernel_manager: RemoteKernelManager, proxy_config: dict):
         """Initialize the proxy."""

enterprise_gateway/services/processproxies/distributed.py

@@ -10,7 +10,7 @@
 import signal
 from socket import gethostbyname
 from subprocess import STDOUT
-from typing import Any
+from typing import Any, ClassVar
 
 from ..kernels.remotemanager import RemoteKernelManager
 from .processproxy import BaseProcessProxyABC, RemoteProcessProxy
@@ -24,8 +24,8 @@
 class TrackKernelOnHost:
     """A class for tracking a kernel on a host."""
 
-    _host_kernels = {}
-    _kernel_host_mapping = {}
+    _host_kernels: ClassVar = {}
+    _kernel_host_mapping: ClassVar = {}
 
     def add_kernel_id(self, host: str, kernel_id: str) -> None:
         """Add a kernel to a host."""

enterprise_gateway/services/processproxies/yarn.py

@@ -11,7 +11,7 @@
 import signal
 import socket
 import time
-from typing import Any
+from typing import Any, ClassVar
 
 from jupyter_client import localinterfaces
 from yarn_api_client.base import Response
@@ -43,8 +43,8 @@ class YarnClusterProcessProxy(RemoteProcessProxy):
     Kernel lifecycle management for YARN clusters.
     """
 
-    initial_states = {"NEW", "SUBMITTED", "ACCEPTED", "RUNNING"}
-    final_states = {"FINISHED", "KILLED", "FAILED"}
+    initial_states: ClassVar = {"NEW", "SUBMITTED", "ACCEPTED", "RUNNING"}
+    final_states: ClassVar = {"FINISHED", "KILLED", "FAILED"}
 
     def __init__(self, kernel_manager: RemoteKernelManager, proxy_config: dict):
         """Initialize the proxy."""
@@ -456,7 +456,7 @@ def _get_application_id(self, ignore_final_states: bool = False) -> str:
         if not self.application_id:
             app = self._query_app_by_name(self.kernel_id)
             state_condition = True
-            if type(app) is dict:
+            if isinstance(app, dict):
                 state = app.get("state")
                 self.last_known_state = state
 
@@ -518,7 +518,11 @@ def _query_app_by_name(self, kernel_id: str) -> dict:
             )
         else:
             data = response.data
-            if type(data) is dict and type(data.get("apps")) is dict and "app" in data.get("apps"):
+            if (
+                isinstance(data, dict)
+                and isinstance(data.get("apps"), dict)
+                and "app" in data.get("apps")
+            ):
                 for app in data["apps"]["app"]:
                     if app.get("name", "").find(kernel_id) >= 0 and app.get("id") > top_most_app_id:
                         target_app = app
@@ -540,7 +544,7 @@ def _query_app_by_id(self, app_id: str) -> dict:
             )
         else:
             data = response.data
-            if type(data) is dict and "app" in data:
+            if isinstance(data, dict) and "app" in data:
                 app = data["app"]
 
         return app

etc/kernel-launchers/R/scripts/server_listener.py

@@ -67,8 +67,7 @@ def return_connection_info(
     response_parts = response_addr.split(":")
     if len(response_parts) != 2:
         logger.error(
-            f"Invalid format for response address '{response_addr}'. "
-            "Assuming 'pull' mode..."
+            f"Invalid format for response address '{response_addr}'. Assuming 'pull' mode..."
         )
         return