diff --git a/dagster_ray/__init__.py b/dagster_ray/__init__.py
index ec62ee3f..bdfd6582 100644
--- a/dagster_ray/__init__.py
+++ b/dagster_ray/__init__.py
@@ -9,11 +9,11 @@
 
 
 __all__ = [
+    "LocalRay",
+    "PipesRayJobClient",
+    "PipesRayJobMessageReader",
+    "RayIOManager",
     "RayResource",
     "RayRunLauncher",
-    "RayIOManager",
     "ray_executor",
-    "PipesRayJobMessageReader",
-    "PipesRayJobClient",
-    "LocalRay",
 ]
diff --git a/dagster_ray/_base/resources.py b/dagster_ray/_base/resources.py
index 4e55ac97..59405d2d 100644
--- a/dagster_ray/_base/resources.py
+++ b/dagster_ray/_base/resources.py
@@ -1,12 +1,11 @@
+from __future__ import annotations
+
 import uuid
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional, Union, cast
+from typing import TYPE_CHECKING, Union, cast
 
 from dagster import AssetExecutionContext, ConfigurableResource, InitResourceContext, OpExecutionContext
 from pydantic import Field, PrivateAttr
-
-# yes, `python-client` is actually the KubeRay package name
-# https://github.com/ray-project/kuberay/issues/2078
 from requests.exceptions import ConnectionError
 from tenacity import retry, retry_if_exception_type, stop_after_delay
 from typing_extensions import TypeAlias
@@ -15,42 +14,46 @@
 from dagster_ray.config import RayDataExecutionOptions
 
 if TYPE_CHECKING:
-    from ray._private.worker import BaseContext as RayBaseContext  # noqa
+    from ray._private.worker import BaseContext as RayBaseContext
 
 OpOrAssetExecutionContext: TypeAlias = Union[OpExecutionContext, AssetExecutionContext]
 
 
 class BaseRayResource(ConfigurableResource, ABC):
-    """
-    Base class for Ray Resources.
+    """Base class for Ray Resources.
     Defines the common interface and some utility methods.
     """
 
     data_execution_options: RayDataExecutionOptions = Field(default_factory=RayDataExecutionOptions)
     redis_port: int = Field(
-        default=10001, description="Redis port for connection. Make sure to match with the actual available port."
+        default=10001,
+        description="Redis port for connection. Make sure to match with the actual available port.",
     )
     dashboard_port: int = Field(
-        default=8265, description="Dashboard port for connection. Make sure to match with the actual available port."
+        default=8265,
+        description="Dashboard port for connection. Make sure to match with the actual available port.",
     )
 
-    _context: Optional["RayBaseContext"] = PrivateAttr()
+    _context: RayBaseContext | None = PrivateAttr()
 
     def setup_for_execution(self, context: InitResourceContext) -> None:
-        raise NotImplementedError(
+        msg = (
             "This is an abstract resource, it's not meant to be provided directly. "
             "Use a backend-specific resource instead."
         )
+        raise NotImplementedError(
+            msg,
+        )
 
     @property
-    def context(self) -> "RayBaseContext":
+    def context(self) -> RayBaseContext:
         assert self._context is not None, "RayClusterResource not initialized"
         return self._context
 
     @property
     @abstractmethod
     def host(self) -> str:
-        raise NotImplementedError()
+        raise NotImplementedError
 
     @property
     def ray_address(self) -> str:
@@ -62,8 +65,7 @@ def dashboard_url(self) -> str:
 
     @property
     def runtime_job_id(self) -> str:
-        """
-        Returns the Ray Job ID for the current job which was created with `ray.init()`.
+        """Returns the Ray Job ID for the current job which was created with `ray.init()`.
         :return:
         """
         import ray
@@ -71,7 +73,7 @@ def runtime_job_id(self) -> str:
         return ray.get_runtime_context().get_job_id()
 
     @retry(stop=stop_after_delay(120), retry=retry_if_exception_type(ConnectionError), reraise=True)
-    def init_ray(self, context: Union[OpOrAssetExecutionContext, InitResourceContext]) -> "RayBaseContext":
+    def init_ray(self, context: OpOrAssetExecutionContext | InitResourceContext) -> RayBaseContext:
         assert context.log is not None
 
         import ray
@@ -84,8 +86,7 @@ def init_ray(self, context: Union[OpOrAssetExecutionContext, InitResourceContext
         return cast("RayBaseContext", self._context)
 
     def get_dagster_tags(self, context: InitResourceContext) -> dict[str, str]:
-        tags = get_dagster_tags(context)
-        return tags
+        return get_dagster_tags(context)
 
     def _get_step_key(self, context: InitResourceContext) -> str:
         # just return a random string
diff --git a/dagster_ray/_base/utils.py b/dagster_ray/_base/utils.py
index 8b780c4d..4b9ff6b6 100644
--- a/dagster_ray/_base/utils.py
+++ b/dagster_ray/_base/utils.py
@@ -1,15 +1,16 @@
-import os
-from typing import Union, cast
+from __future__ import annotations
 
-from dagster import AssetExecutionContext, InitResourceContext, OpExecutionContext
+import os
+from typing import TYPE_CHECKING, cast
 
 from dagster_ray._base.constants import DEFAULT_DEPLOYMENT_NAME
 
+if TYPE_CHECKING:
+    from dagster import AssetExecutionContext, InitResourceContext, OpExecutionContext
+
 
-def get_dagster_tags(context: Union[InitResourceContext, OpExecutionContext, AssetExecutionContext]) -> dict[str, str]:
-    """
-    Returns a dictionary with common Dagster tags.
-    """
+def get_dagster_tags(context: InitResourceContext | OpExecutionContext | AssetExecutionContext) -> dict[str, str]:
+    """Returns a dictionary with common Dagster tags."""
     assert context.dagster_run is not None
 
     labels = {
diff --git a/dagster_ray/config.py b/dagster_ray/config.py
index fb3fe6d6..201a856f 100644
--- a/dagster_ray/config.py
+++ b/dagster_ray/config.py
@@ -1,11 +1,13 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from dagster import Config
 from pydantic import Field
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
 USER_DEFINED_RAY_KEY = "dagster-ray/config"
 
 
@@ -20,8 +22,7 @@ class RayExecutionConfig(Config):
     def from_tags(cls, tags: Mapping[str, str]) -> RayExecutionConfig:
         if USER_DEFINED_RAY_KEY in tags:
             return cls.parse_raw(tags[USER_DEFINED_RAY_KEY])
-        else:
-            return cls()
+        return cls()
 
 
 class RayJobSubmissionClientConfig(Config):
@@ -38,7 +39,8 @@ class RayJobSubmissionClientConfig(Config):
             for cases like authentication to a remote cluster.""",
     )
     cookies: dict[str, Any] | None = Field(
-        default=None, description="Cookies to use when sending requests to the HTTP job server."
+        default=None,
+        description="Cookies to use when sending requests to the HTTP job server.",
     )
 
 
@@ -56,7 +58,7 @@ class RayDataExecutionOptions(Config):
     verbose_progress: bool = True
     use_polars: bool = True
 
-    def apply(self):
+    def apply(self) -> None:
         import ray
         from ray.data import ExecutionResources
 
@@ -71,11 +73,11 @@ def apply(self):
         ctx.verbose_progress = self.verbose_progress
         ctx.use_polars = self.use_polars
 
-    def apply_remote(self):
+    def apply_remote(self) -> None:
         import ray
 
         @ray.remote
-        def apply():
+        def apply() -> None:
             self.apply()
 
         ray.get(apply.remote())
diff --git a/dagster_ray/executor.py b/dagster_ray/executor.py
index 349b32e1..a125bfe1 100644
--- a/dagster_ray/executor.py
+++ b/dagster_ray/executor.py
@@ -1,4 +1,5 @@
-from collections.abc import Iterator
+from __future__ import annotations
+
 from typing import TYPE_CHECKING, Any, Optional, cast
 
 import dagster
@@ -13,8 +14,6 @@
 from dagster._core.events import DagsterEvent, EngineEventData
 from dagster._core.execution.retries import RetryMode, get_retries_config
 from dagster._core.execution.tags import get_tag_concurrency_limits_config
-from dagster._core.executor.base import Executor
-from dagster._core.executor.init import InitExecutorContext
 from dagster._core.executor.step_delegating import (
     CheckStepHealthResult,
     StepDelegatingExecutor,
@@ -32,15 +31,20 @@
 from dagster_ray.utils import resolve_env_vars_list
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from dagster._core.executor.base import Executor
+    from dagster._core.executor.init import InitExecutorContext
     from ray.job_submission import JobSubmissionClient
 
 
 class RayExecutorConfig(RayExecutionConfig, RayJobSubmissionClientConfig):
-    env_vars: Optional[list[str]] = Field(
+    env_vars: list[str] | None = Field(
         default=None,
-        description="A list of environment variables to inject into the Job. Each can be of the form KEY=VALUE or just KEY (in which case the value will be pulled from the current process).",
+        description="""A list of environment variables to inject into the Job. Each can be of the form KEY=VALUE
+            or just KEY (in which case the value will be pulled from the current process).""",
     )
-    address: Optional[str] = Field(default=None, description="The address of the Ray cluster to connect to.")  # type: ignore
+    address: str | None = Field(default=None, description="The address of the Ray cluster to connect to.")  # type: ignore
     # sorry for the long name, but it has to be very clear what this is doing
     inherit_job_submission_client_from_ray_run_launcher: bool = True
 
@@ -71,13 +75,17 @@ def ray_executor(init_context: InitExecutorContext) -> Executor:
     ray_cfg = RayExecutorConfig(**exc_cfg["ray"])  # type: ignore
 
     if ray_cfg.inherit_job_submission_client_from_ray_run_launcher and isinstance(
-        init_context.instance.run_launcher, RayRunLauncher
+        init_context.instance.run_launcher,
+        RayRunLauncher,
     ):
         # TODO: some RunLauncher config values can be automatically passed to the executor
         client = init_context.instance.run_launcher.client
     else:
         client = JobSubmissionClient(
-            ray_cfg.address, metadata=ray_cfg.metadata, headers=ray_cfg.headers, cookies=ray_cfg.cookies
+            ray_cfg.address,
+            metadata=ray_cfg.metadata,
+            headers=ray_cfg.headers,
+            cookies=ray_cfg.cookies,
         )
 
     return StepDelegatingExecutor(
@@ -99,19 +107,19 @@ def ray_executor(init_context: InitExecutorContext) -> Executor:
 
 class RayStepHandler(StepHandler):
     @property
-    def name(self):
+    def name(self) -> str:
         return "RayStepHandler"
 
     def __init__(
         self,
-        client: "JobSubmissionClient",
-        env_vars: Optional[list[str]],
-        runtime_env: Optional[dict[str, Any]],
-        num_cpus: Optional[float],
-        num_gpus: Optional[float],
-        memory: Optional[int],
-        resources: Optional[dict[str, float]],
-    ):
+        client: JobSubmissionClient,
+        env_vars: list[str] | None,
+        runtime_env: dict[str, Any] | None,
+        num_cpus: float | None,
+        num_gpus: float | None,
+        memory: int | None,
+        resources: dict[str, float] | None,
+    ) -> None:
         super().__init__()
 
         self.client = client
@@ -127,7 +135,7 @@ def _get_step_key(self, step_handler_context: StepHandlerContext) -> str:
         assert len(step_keys_to_execute) == 1, "Launching multiple steps is not currently supported"
         return step_keys_to_execute[0]
 
-    def _get_ray_job_submission_id(self, step_handler_context: StepHandlerContext):
+    def _get_ray_job_submission_id(self, step_handler_context: StepHandlerContext) -> str:
         step_key = self._get_step_key(step_handler_context)
 
         name_key = get_k8s_object_name(
@@ -195,7 +203,7 @@ def launch_step(self, step_handler_context: StepHandlerContext) -> Iterator[Dags
 
         self.client.submit_job(
             entrypoint=" ".join(
-                step_handler_context.execute_step_args.get_command_args(skip_serialized_namedtuple=True)
+                step_handler_context.execute_step_args.get_command_args(skip_serialized_namedtuple=True),
             ),
             submission_id=submission_id,
             metadata=labels,
@@ -217,7 +225,7 @@ def check_step_health(self, step_handler_context: StepHandlerContext) -> CheckSt
             status = self.client.get_job_status(submission_id)
         except RuntimeError:
             return CheckStepHealthResult.unhealthy(
-                reason=f"Ray job {submission_id} for step {step_key} could not be found."
+                reason=f"Ray job {submission_id} for step {step_key} could not be found.",
             )
 
         if status == JobStatus.FAILED:
diff --git a/dagster_ray/io_manager.py b/dagster_ray/io_manager.py
index 5809f8be..863d3bfe 100644
--- a/dagster_ray/io_manager.py
+++ b/dagster_ray/io_manager.py
@@ -1,9 +1,10 @@
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING
 
 from dagster import ConfigurableIOManager, InputContext, OutputContext
 
 DAGSTER_RAY_OBJECT_MAP_NAME = "DagsterRayObjectMap"
 DAGSTER_RAY_NAMESPACE = "dagster-ray"
+from __future__ import annotations
 
 # we need to create a global Ray actor which will store all the refs to all objcets
 
@@ -15,23 +16,23 @@ class RayObjectMap:
     # TODO: implement some eventual cleanup mechanism
     # idea: save creation timestamp and periodically check for old refs
     # or add some integration with the RunLauncher/Executor
-    def __init__(self):
+    def __init__(self) -> None:
         self._object_map: dict[str, ray.ObjectRef] = {}
 
-    def set(self, key: str, ref: "ray.ObjectRef"):
+    def set(self, key: str, ref: ray.ObjectRef) -> None:
         self._object_map[key] = ref
 
-    def get(self, key: str) -> Optional["ray.ObjectRef"]:
+    def get(self, key: str) -> ray.ObjectRef | None:
         return self._object_map.get(key)
 
-    def delete(self, key: str):
+    def delete(self, key: str) -> None:
         if key in self._object_map:
             del self._object_map[key]
 
     def keys(self):
         return self._object_map.keys()
 
-    def ping(self):
+    def ping(self) -> str:
         return "pong"
 
     @staticmethod
@@ -59,9 +60,9 @@ def get_or_create():
 
 
 class RayIOManager(ConfigurableIOManager):
-    address: Optional[str] = None
+    address: str | None = None
 
-    def handle_output(self, context: OutputContext, obj):
+    def handle_output(self, context: OutputContext, obj) -> None:
         import ray
 
         if self.address:  # TODO: should this really be done here?
@@ -95,10 +96,9 @@ def load_input(self, context: InputContext):
             storage_keys = self._get_multiple_keys(context)
             refs = [object_map.get.remote(key) for key in storage_keys.values()]  # type: ignore
             values = ray.get(refs)
-            return {partition_key: value for partition_key, value in zip(storage_keys.keys(), values)}
+            return dict(zip(storage_keys.keys(), values))
 
-        else:
-            storage_key = self._get_single_key(context)
+        storage_key = self._get_single_key(context)
 
         context.log.debug(f"[RayIOManager] Loading object with key {storage_key}")
 
@@ -108,7 +108,7 @@ def load_input(self, context: InputContext):
 
         return ray.get(ref)
 
-    def _get_single_key(self, context: Union[InputContext, OutputContext]) -> str:
+    def _get_single_key(self, context: InputContext | OutputContext) -> str:
         identifier = context.get_identifier() if not context.has_asset_key else context.get_asset_identifier()
         return "/".join(identifier)
 
@@ -117,9 +117,9 @@ def _get_multiple_keys(self, context: InputContext) -> dict[str, str]:
             asset_path = list(context.asset_key.path)
 
             return {
-                partition_key: "/".join(asset_path + [partition_key]) for partition_key in context.asset_partition_keys
+                partition_key: "/".join([*asset_path, partition_key]) for partition_key in context.asset_partition_keys
             }
-        else:
-            raise RuntimeError(
-                "[RayIOManager] This method can only be called with an InputContext that has multiple partitions"
-            )
+        msg = "[RayIOManager] This method can only be called with an InputContext that has multiple partitions"
+        raise RuntimeError(
+            msg,
+        )
diff --git a/dagster_ray/kuberay/__init__.py b/dagster_ray/kuberay/__init__.py
index 465db0d0..d08a4e1e 100644
--- a/dagster_ray/kuberay/__init__.py
+++ b/dagster_ray/kuberay/__init__.py
@@ -7,12 +7,12 @@
 
 __all__ = [
     "KubeRayCluster",
-    "RayClusterConfig",
-    "RayClusterClientResource",
     "PipesKubeRayJobClient",
+    "RayClusterClientResource",
+    "RayClusterConfig",
     "cleanup_kuberay_clusters",
-    "delete_kuberay_clusters",
+    "cleanup_kuberay_clusters_daily",
     "cleanup_kuberay_clusters_op",
+    "delete_kuberay_clusters",
     "delete_kuberay_clusters_op",
-    "cleanup_kuberay_clusters_daily",
 ]
diff --git a/dagster_ray/kuberay/client/base.py b/dagster_ray/kuberay/client/base.py
index 0ba721bf..aa7d5b7a 100644
--- a/dagster_ray/kuberay/client/base.py
+++ b/dagster_ray/kuberay/client/base.py
@@ -1,12 +1,14 @@
+from __future__ import annotations
+
 import time
-from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 if TYPE_CHECKING:
     from kubernetes import client
     from kubernetes.client.models.v1_endpoints import V1Endpoints
 
 
-def load_kubeconfig(context: Optional[str] = None, config_file: Optional[str] = None) -> Any:
+def load_kubeconfig(context: str | None = None, config_file: str | None = None) -> Any:
     from kubernetes import config
 
     try:
@@ -28,8 +30,8 @@ def __init__(
         version: str,
         kind: str,
         plural: str,
-        api_client: Optional["client.ApiClient"] = None,
-    ):
+        api_client: client.ApiClient | None = None,
+    ) -> None:
         from kubernetes import client
 
         self.group = group
@@ -40,7 +42,9 @@ def __init__(
         self._api = client.CustomObjectsApi(api_client=api_client)
         self._core_v1_api = client.CoreV1Api(api_client=api_client)
 
-    def wait_for_service_endpoints(self, service_name: str, namespace: str, poll_interval: int = 5, timeout: int = 600):
+    def wait_for_service_endpoints(
+        self, service_name: str, namespace: str, poll_interval: int = 5, timeout: int = 600
+    ) -> None:
         from kubernetes.client import ApiException
 
         start_time = time.time()
@@ -60,8 +64,9 @@ def wait_for_service_endpoints(self, service_name: str, namespace: str, poll_int
 
             elapsed_time = time.time() - start_time
             if elapsed_time > timeout:
+                msg = f"Timed out waiting for endpoints for service {service_name} in namespace {namespace}"
                 raise TimeoutError(
-                    f"Timed out waiting for endpoints for service {service_name} in namespace {namespace}"
+                    msg,
                 )
 
             time.sleep(poll_interval)
@@ -83,11 +88,11 @@ def get_status(self, name: str, namespace: str, timeout: int = 60, poll_interval
 
             if resource.get("status"):
                 return resource["status"]
-            else:
-                time.sleep(poll_interval)
-                timeout -= poll_interval
+            time.sleep(poll_interval)
+            timeout -= poll_interval
 
-        raise TimeoutError(f"Timed out waiting for status of {self.kind} {name} in namespace {namespace}")
+        msg = f"Timed out waiting for status of {self.kind} {name} in namespace {namespace}"
+        raise TimeoutError(msg)
 
     def list(self, namespace: str, label_selector: str = "", async_req: bool = False) -> dict[str, Any]:
         from kubernetes.client import ApiException
@@ -103,8 +108,7 @@ def list(self, namespace: str, label_selector: str = "", async_req: bool = False
             )
             if "items" in resource:
                 return resource
-            else:
-                return {}
+            return {}
         except ApiException as e:
             if e.status == 404:
                 return {}
diff --git a/dagster_ray/kuberay/client/raycluster/client.py b/dagster_ray/kuberay/client/raycluster/client.py
index abbef491..6e7856cd 100644
--- a/dagster_ray/kuberay/client/raycluster/client.py
+++ b/dagster_ray/kuberay/client/raycluster/client.py
@@ -1,16 +1,15 @@
+from __future__ import annotations
+
 import logging
 import socket
 import subprocess
 import threading
 import time
-from collections.abc import Iterator
 from contextlib import contextmanager
-from io import FileIO
 from queue import Queue
 from typing import (
     TYPE_CHECKING,
     Any,
-    Optional,
     TypedDict,
     cast,
 )
@@ -22,6 +21,9 @@
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from io import FileIO
+
     from kubernetes.client import ApiClient
     from ray.job_submission import JobSubmissionClient
 
@@ -32,7 +34,7 @@
 KIND = "RayCluster"
 
 
-def enqueue_output(out: FileIO, queue: Queue, should_stop):
+def enqueue_output(out: FileIO, queue: Queue, should_stop) -> None:
     for line in iter(out.readline, b""):
         if should_stop():
             return
@@ -78,9 +80,9 @@ class RayClusterStatus(TypedDict):
 class RayClusterClient(BaseKubeRayClient[RayClusterStatus]):
     def __init__(
         self,
-        config_file: Optional[str] = None,
-        context: Optional[str] = None,
-        api_client: Optional["ApiClient"] = None,
+        config_file: str | None = None,
+        context: str | None = None,
+        api_client: ApiClient | None = None,
     ) -> None:
         super().__init__(group=GROUP, version=VERSION, kind=KIND, plural=PLURAL, api_client=api_client)
 
@@ -94,7 +96,7 @@ def wait_until_ready(
         name: str,
         namespace: str,
         timeout: int,
-        image: Optional[str] = None,
+        image: str | None = None,
     ) -> tuple[str, dict[str, str]]:
         from kubernetes import watch
 
@@ -126,8 +128,9 @@ def wait_until_ready(
             status: RayClusterStatus = item["status"]
 
             if status.get("state") == "failed":
-                raise Exception(
-                    f"RayCluster {namespace}/{name} failed to start. Reason:\n{status.get('reason')}\nMore details: `kubectl -n {namespace} describe RayCluster {name}`"
+                msg = f"RayCluster {namespace}/{name} failed to start. Reason:\n{status.get('reason')}\nMore details: `kubectl -n {namespace} describe RayCluster {name}`"  # noqa: E501
+                raise RuntimeError(
+                    msg,
                 )
 
             if (
@@ -137,22 +140,23 @@ def wait_until_ready(
                 and status.get("head")
                 and status.get("endpoints", {}).get("dashboard")
             ):
-                if image is not None:
-                    if (
-                        item.get("spec")  # type: ignore
-                        and item["spec"]["headGroupSpec"]["template"]["spec"]["containers"][0]["image"]  # type: ignore
-                        != image
-                    ):
-                        continue
+                if image is not None and (
+                    item.get("spec")  # type: ignore
+                    and item["spec"]["headGroupSpec"]["template"]["spec"]["containers"][0]["image"]  # type: ignore
+                    != image
+                ):
+                    continue
                 w.stop()
                 logger.debug(f"RayCluster {namespace}/{name} is ready!")
                 return status["head"]["serviceIP"], status["endpoints"]  # type: ignore
 
             if time.time() - start_time > timeout:
                 w.stop()
-                raise TimeoutError(f"Timed out waiting for RayCluster {namespace}/{name} to be ready. Status: {status}")
+                msg = f"Timed out waiting for RayCluster {namespace}/{name} to be ready. Status: {status}"
+                raise TimeoutError(msg)
 
-        raise Exception("This code should be unreachable")
+        msg = "This code should be unreachable"
+        raise Exception(msg)
 
     @contextmanager
     def port_forward(
@@ -162,8 +166,7 @@ def port_forward(
         local_dashboard_port: int = 8265,
         local_gcs_port: int = 10001,
     ) -> Iterator[tuple[int, int]]:
-        """
-        Port forwards the Ray dashboard and GCS ports to localhost.
+        """Port forwards the Ray dashboard and GCS ports to localhost.
         Use 0 for local_dashboard_port and local_gcs_port to get random available ports.
         Returns the ports that the dashboard and GCS are forwarded to.
         """
@@ -176,7 +179,8 @@ def port_forward(
         service = f"{name}-head-svc"
 
         if not self.get_status(name, namespace):
-            raise RuntimeError(f"RayCluster {name} does not exist in namespace {namespace}")
+            msg = f"RayCluster {name} does not exist in namespace {namespace}"
+            raise RuntimeError(msg)
 
         self.wait_for_service_endpoints(service_name=service, namespace=namespace)
 
@@ -217,8 +221,9 @@ def should_stop():
 
             while True:
                 if process.poll() is not None:
+                    msg = f"port-forwarding command: `{' '.join(cmd)}` failed. Most likely ports {local_dashboard_port} or {local_gcs_port} are  already in use, or Kubernetes service {name}-head-svc does not exist in namespace {namespace}."  # noqa: E501
                     raise RuntimeError(
-                        f"port-forwarding command: `{' '.join(cmd)}` failed. Most likely ports {local_dashboard_port} or {local_gcs_port} are  already in use, or Kubernetes service {name}-head-svc does not exist in namespace {namespace}."
+                        msg,
                     )
 
                 line = queue.get()
@@ -227,7 +232,7 @@ def should_stop():
                     break
 
             logger.info(
-                f"Connecting to {namespace}/{name} via port-forwarding for ports {local_dashboard_port} and {local_gcs_port}..."
+                f"Connecting to {namespace}/{name} via port-forwarding for ports {local_dashboard_port} and {local_gcs_port}...",  # noqa: E501
             )
 
             yield local_dashboard_port, local_gcs_port
@@ -241,14 +246,19 @@ def should_stop():
 
     @contextmanager
     def job_submission_client(
-        self, name: str, namespace: str, port_forward: bool = False, timeout: int = 60
-    ) -> Iterator["JobSubmissionClient"]:
-        """
-        Returns a JobSubmissionClient object that can be used to interact with Ray jobs running in the KubeRay cluster.
-        If port_forward is True, it will port forward the dashboard and GCS ports to localhost, and should be used in a context manager.
-        If port_forward is False, the client will connect to the dashboard directly (assuming the dashboard is accessible from the host).
+        self,
+        name: str,
+        namespace: str,
+        port_forward: bool = False,
+        timeout: int = 60,
+    ) -> Iterator[JobSubmissionClient]:
+        """Returns a JobSubmissionClient object that can be used to interact with Ray jobs running
+            in the KubeRay cluster.
+        If port_forward is True, it will port forward the dashboard and GCS ports to localhost,
+            and should be used in a context manager.
+        If port_forward is False, the client will connect to the dashboard directly
+            (assuming the dashboard is accessible from the host).
         """
-
         from ray.job_submission import JobSubmissionClient
 
         if not port_forward:
diff --git a/dagster_ray/kuberay/client/rayjob/client.py b/dagster_ray/kuberay/client/rayjob/client.py
index d018b7b8..049edbdb 100644
--- a/dagster_ray/kuberay/client/rayjob/client.py
+++ b/dagster_ray/kuberay/client/rayjob/client.py
@@ -1,7 +1,8 @@
+from __future__ import annotations
+
 import logging
 import time
-from collections.abc import Iterator
-from typing import TYPE_CHECKING, Literal, Optional, TypedDict
+from typing import TYPE_CHECKING, Literal, TypedDict
 
 from typing_extensions import NotRequired
 
@@ -9,6 +10,8 @@
 from dagster_ray.kuberay.client.raycluster import RayClusterClient, RayClusterStatus
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
+
     from kubernetes.client import ApiClient
 
 GROUP = "ray.io"
@@ -35,9 +38,9 @@ class RayJobStatus(TypedDict):
 class RayJobClient(BaseKubeRayClient[RayJobStatus]):
     def __init__(
         self,
-        config_file: Optional[str] = None,
-        context: Optional[str] = None,
-        api_client: Optional["ApiClient"] = None,
+        config_file: str | None = None,
+        context: str | None = None,
+        api_client: ApiClient | None = None,
     ) -> None:
         # this call must happen BEFORE creating K8s apis
         load_kubeconfig(config_file=config_file, context=context)
@@ -73,8 +76,9 @@ def wait_until_running(
 
             if status in ["Running", "Complete"]:
                 break
-            elif status == "Failed":
-                raise RuntimeError(f"RayJob {namespace}/{name} deployment failed. Status:\n{status}")
+            if status == "Failed":
+                msg = f"RayJob {namespace}/{name} deployment failed. Status:\n{status}"
+                raise RuntimeError(msg)
 
             if time.time() - start_time > timeout:
                 if terminate_on_timeout:
@@ -83,11 +87,12 @@ def wait_until_running(
                         self.terminate(name, namespace, port_forward=port_forward)
                     except Exception as e:
                         logger.warning(
-                            f"Failed to gracefully terminate RayJob {namespace}/{name}: {e}, will delete it instead."
+                            f"Failed to gracefully terminate RayJob {namespace}/{name}: {e}, will delete it instead.",
                         )
                         self.delete(name, namespace)
 
-                raise TimeoutError(f"Timed out waiting for RayJob {namespace}/{name} to start. Status:\n{status}")
+                msg = f"Timed out waiting for RayJob {namespace}/{name} to start. Status:\n{status}"
+                raise TimeoutError(msg)
 
             time.sleep(poll_interval)
 
@@ -98,7 +103,8 @@ def wait_until_running(
                 break
 
             if time.time() - start_time > timeout:
-                raise TimeoutError(f"Timed out waiting for RayJob {namespace}/{name} to start. Status:\n{status}")
+                msg = f"Timed out waiting for RayJob {namespace}/{name} to start. Status:\n{status}"
+                raise TimeoutError(msg)
 
             time.sleep(poll_interval)
 
@@ -110,7 +116,7 @@ def _wait_for_job_submission(
         namespace: str,
         timeout: int = 600,
         poll_interval: int = 10,
-    ):
+    ) -> None:
         start_time = time.time()
 
         while True:
@@ -118,12 +124,12 @@ def _wait_for_job_submission(
             if status.get("jobDeploymentStatus") in ["Complete", "Failed"]:
                 return
 
-            if (job_status := status.get("jobStatus")) is not None:
-                if job_status != "PENDING":
-                    return
+            if (job_status := status.get("jobStatus")) is not None and job_status != "PENDING":
+                return
 
             if time.time() - start_time > timeout:
-                raise TimeoutError(f"Timed out waiting for job {name} to start")
+                msg = f"Timed out waiting for job {name} to start"
+                raise TimeoutError(msg)
 
             logger.debug(f"RayJob {namespace}/{name} deployment status is {job_status}, waiting for it to start...")
 
@@ -132,18 +138,26 @@ def _wait_for_job_submission(
     def get_job_logs(self, name: str, namespace: str, timeout: int = 60 * 60, port_forward: bool = False) -> str:
         self._wait_for_job_submission(name, namespace, timeout=timeout)
         with self.ray_cluster_client.job_submission_client(
-            name=self.get_ray_cluster_name(name, namespace), namespace=namespace, port_forward=port_forward
+            name=self.get_ray_cluster_name(name, namespace),
+            namespace=namespace,
+            port_forward=port_forward,
         ) as job_submission_client:
             return job_submission_client.get_job_logs(job_id=self.get_job_sumission_id(name, namespace))
 
     def tail_job_logs(
-        self, name: str, namespace: str, timeout: int = 60 * 60, port_forward: bool = False
+        self,
+        name: str,
+        namespace: str,
+        timeout: int = 60 * 60,
+        port_forward: bool = False,
     ) -> Iterator[str]:
         import asyncio
 
         self._wait_for_job_submission(name, namespace, timeout=timeout)
         with self.ray_cluster_client.job_submission_client(
-            name=self.get_ray_cluster_name(name, namespace), namespace=namespace, port_forward=port_forward
+            name=self.get_ray_cluster_name(name, namespace),
+            namespace=namespace,
+            port_forward=port_forward,
         ) as job_submission_client:
             async_tailer = job_submission_client.tail_job_logs(job_id=self.get_job_sumission_id(name, namespace))
 
@@ -158,11 +172,11 @@ def tail_logs() -> Iterator[str]:
             yield from tail_logs()
 
     def terminate(self, name: str, namespace: str, port_forward: bool = False) -> bool:
-        """
-        Unlike the .delete method, this won't remove the Kubernetes object, but will instead stop the Ray Job.
-        """
+        """Unlike the .delete method, this won't remove the Kubernetes object, but will instead stop the Ray Job."""
         with self.ray_cluster_client.job_submission_client(
-            name=self.get_ray_cluster_name(name, namespace), namespace=namespace, port_forward=port_forward
+            name=self.get_ray_cluster_name(name, namespace),
+            namespace=namespace,
+            port_forward=port_forward,
         ) as job_submission_client:
             job_id = self.get_job_sumission_id(name, namespace)
 
@@ -177,7 +191,7 @@ def terminate(self, name: str, namespace: str, port_forward: bool = False) -> bo
                         break
 
                 logger.debug(
-                    f"Trying to terminate job {name}, but it wasn't submitted yet. Waiting for it to be submitted..."
+                    f"Trying to terminate job {name}, but it wasn't submitted yet. Waiting for it to be submitted...",
                 )
                 time.sleep(10)
 
diff --git a/dagster_ray/kuberay/configs.py b/dagster_ray/kuberay/configs.py
index ba30c4fc..1e1b3036 100644
--- a/dagster_ray/kuberay/configs.py
+++ b/dagster_ray/kuberay/configs.py
@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import os
-from typing import Any, Literal, Optional
+from typing import Any, Literal
 
 from dagster import Config
 from pydantic import Field
@@ -61,7 +63,7 @@
                         "volumeMounts": [{"mountPath": "/tmp/ray", "name": "ray-logs"}],
                         "name": "worker",
                         "imagePullPolicy": "Always",
-                    }
+                    },
                 ],
                 "volumes": [
                     {"name": "ray-logs", "emptyDir": {}},
@@ -71,12 +73,12 @@
                 "nodeSelector": {},
             },
         },
-    }
+    },
 ]
 
 
 class RayClusterConfig(Config):
-    image: Optional[str] = None
+    image: str | None = None
     namespace: str = "ray"
     enable_in_tree_autoscaling: bool = False
     autoscaler_options: dict[str, Any] = DEFAULT_AUTOSCALER_OPTIONS  # TODO: add a dedicated Config type
@@ -89,11 +91,11 @@ class RayJobConfig(Config):
     entrypoint_memory: float
     entrypoint_num_gpus: int
     suspend: bool = False
-    annotations: Optional[dict[str, str]] = None
-    labels: Optional[dict[str, str]] = None
+    annotations: dict[str, str] | None = None
+    labels: dict[str, str] | None = None
     shutdown_after_job_finishes: bool = True
     ttl_seconds_after_finished: int = 60 * 10  # 10 minutes
     active_deadline_seconds: int = 60 * 60 * 24  # 24 hours
     submission_mode: Literal["K8sJobMode", "HTTPMode"] = "K8sJobMode"
-    runtime_env_yaml: Optional[str] = None
+    runtime_env_yaml: str | None = None
     cluster: RayClusterConfig = Field(default_factory=RayClusterConfig)
diff --git a/dagster_ray/kuberay/jobs.py b/dagster_ray/kuberay/jobs.py
index 02ad2adf..a70ae8bc 100644
--- a/dagster_ray/kuberay/jobs.py
+++ b/dagster_ray/kuberay/jobs.py
@@ -4,13 +4,14 @@
 
 
 @job(description="Deletes RayCluster resources from Kubernetes", name="delete_kuberay_rayclusters")
-def delete_kuberay_clusters():
+def delete_kuberay_clusters() -> None:
     delete_kuberay_clusters_op()
 
 
 @job(
-    description="Deletes RayCluster resources which do not correspond to any active Dagster Runs in this deployment from Kubernetes",
+    description="""Deletes RayCluster resources which do not correspond to any active Dagster Runs
+    in this deployment from Kubernetes""",
     name="cleanup_kuberay_rayclusters",
 )
-def cleanup_kuberay_clusters():
+def cleanup_kuberay_clusters() -> None:
     cleanup_kuberay_clusters_op()
diff --git a/dagster_ray/kuberay/ops.py b/dagster_ray/kuberay/ops.py
index a0e89a1f..be299a06 100644
--- a/dagster_ray/kuberay/ops.py
+++ b/dagster_ray/kuberay/ops.py
@@ -30,12 +30,14 @@ def delete_kuberay_clusters_op(
 class CleanupKuberayClustersConfig(Config):
     namespace: str = "kuberay"
     label_selector: str = Field(
-        default=f"dagster.io/deployment={DEFAULT_DEPLOYMENT_NAME}", description="Label selector to filter RayClusters"
+        default=f"dagster.io/deployment={DEFAULT_DEPLOYMENT_NAME}",
+        description="Label selector to filter RayClusters",
     )
 
 
 @op(
-    description="Deletes RayCluster resources which do not correspond to any active Dagster Runs in this deployment from Kubernetes",
+    description="""Deletes RayCluster resources which do not correspond to any active Dagster Runs in this deployment
+        from Kubernetes""",
     name="cleanup_kuberay_clusters",
 )
 def cleanup_kuberay_clusters_op(
@@ -49,8 +51,8 @@ def cleanup_kuberay_clusters_op(
                 DagsterRunStatus.STARTED,
                 DagsterRunStatus.QUEUED,
                 DagsterRunStatus.CANCELING,
-            ]
-        )
+            ],
+        ),
     )
 
     clusters = client.client.list(
diff --git a/dagster_ray/kuberay/pipes.py b/dagster_ray/kuberay/pipes.py
index 03b160f8..29782ab8 100644
--- a/dagster_ray/kuberay/pipes.py
+++ b/dagster_ray/kuberay/pipes.py
@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import time
-from typing import TYPE_CHECKING, Any, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Union, cast
 
 import dagster._check as check
 import yaml
@@ -12,9 +14,7 @@
     PipesContextInjector,
     PipesMessageReader,
 )
-from dagster._core.pipes.context import PipesSession
 from dagster._core.pipes.utils import PipesEnvContextInjector, open_pipes_session
-from dagster_pipes import PipesExtras
 from typing_extensions import TypeAlias
 
 from dagster_ray._base.utils import get_dagster_tags
@@ -23,6 +23,8 @@
 from dagster_ray.pipes import PipesRayJobMessageReader, generate_job_id
 
 if TYPE_CHECKING:
+    from dagster._core.pipes.context import PipesSession
+    from dagster_pipes import PipesExtras
     from ray.job_submission import JobSubmissionClient
 
 OpOrAssetExecutionContext: TypeAlias = Union[OpExecutionContext, AssetExecutionContext]
@@ -38,24 +40,25 @@ class PipesKubeRayJobClient(PipesClient, TreatAsResourceParam):
         message_reader (Optional[PipesMessageReader]): A message reader to use to read messages
             from the glue job run. Defaults to :py:class:`PipesRayJobMessageReader`.
         client (Optional[boto3.client]): The Kubernetes API client.
-        forward_termination (bool): Whether to terminate the Ray job when the Dagster process receives a termination signal,
-            or if the startup timeout is reached. Defaults to ``True``.
+        forward_termination (bool): Whether to terminate the Ray job when the Dagster process receives
+            a termination signal, or if the startup timeout is reached. Defaults to ``True``.
         timeout (int): Timeout for various internal interactions with the Kubernetes RayJob.
         poll_interval (int): Interval at which to poll the Kubernetes for status updates.
         port_forward (bool): Whether to use Kubernetes port-forwarding to connect to the KubeRay cluster.
         Is useful when running in a local environment.
+
     """
 
     def __init__(
         self,
-        client: Optional[RayJobClient] = None,
-        context_injector: Optional[PipesContextInjector] = None,
-        message_reader: Optional[PipesMessageReader] = None,
+        client: RayJobClient | None = None,
+        context_injector: PipesContextInjector | None = None,
+        message_reader: PipesMessageReader | None = None,
         forward_termination: bool = True,
         timeout: int = 600,
         poll_interval: int = 5,
         port_forward: bool = False,
-    ):
+    ) -> None:
         self.client: RayJobClient = client or RayJobClient()
 
         self._context_injector = context_injector or PipesEnvContextInjector()
@@ -66,29 +69,29 @@ def __init__(
         self.poll_interval = check.int_param(poll_interval, "poll_interval")
         self.port_forward = check.bool_param(port_forward, "port_forward")
 
-        self._job_submission_client: Optional[JobSubmissionClient] = None
+        self._job_submission_client: JobSubmissionClient | None = None
 
     @property
-    def job_submission_client(self) -> "JobSubmissionClient":
+    def job_submission_client(self) -> JobSubmissionClient:
         if self._job_submission_client is None:
-            raise DagsterInvariantViolationError("JobSubmissionClient is only available inside the run method.")
-        else:
-            return self._job_submission_client
+            msg = "JobSubmissionClient is only available inside the run method."
+            raise DagsterInvariantViolationError(msg)
+        return self._job_submission_client
 
     def run(  # type: ignore
         self,
         *,
         context: OpOrAssetExecutionContext,
         ray_job: dict[str, Any],
-        extras: Optional[PipesExtras] = None,
+        extras: PipesExtras | None = None,
     ) -> PipesClientCompletedInvocation:
-        """
-        Execute a RayJob, enriched with the Pipes protocol.
+        """Execute a RayJob, enriched with the Pipes protocol.
 
         Args:
             context (OpExecutionContext): Current Dagster op or asset context.
             ray_job (Dict[str, Any]): RayJob specification. `API reference <https://ray-project.github.io/kuberay/reference/api/#rayjob>`_.
             extras (Optional[Dict[str, Any]]): Additional information to pass to the Pipes session.
+
         """
         with open_pipes_session(
             context=context,
@@ -117,17 +120,19 @@ def run(  # type: ignore
                 except DagsterExecutionInterruptedError:
                     if self.forward_termination:
                         context.log.warning(
-                            f"[pipes] Dagster process interrupted! Will terminate RayJob {namespace}/{name}."
+                            f"[pipes] Dagster process interrupted! Will terminate RayJob {namespace}/{name}.",
                         )
                         self._terminate(context, start_response)
                     raise
 
     def get_dagster_tags(self, context: OpOrAssetExecutionContext) -> dict[str, str]:
-        tags = get_dagster_tags(context)
-        return tags
+        return get_dagster_tags(context)
 
     def _enrich_ray_job(
-        self, context: OpOrAssetExecutionContext, session: PipesSession, ray_job: dict[str, Any]
+        self,
+        context: OpOrAssetExecutionContext,
+        session: PipesSession,
+        ray_job: dict[str, Any],
     ) -> dict[str, Any]:
         env_vars = session.get_bootstrap_env_vars()
 
@@ -213,12 +218,14 @@ def _wait_for_completion(self, context: OpOrAssetExecutionContext, start_respons
                     context.log.info(f"[pipes] RayJob {namespace}/{name} is complete!")
                     return status
                 elif job_status in ["STOPPED", "FAILED"]:
+                    msg = f"RayJob {namespace}/{name} status is {job_status}. Message:\n{status.get('message')}"
                     raise RuntimeError(
-                        f"RayJob {namespace}/{name} status is {job_status}. Message:\n{status.get('message')}"
+                        msg,
                     )
                 else:
+                    msg = f"RayJob {namespace}/{name} has an unknown status: {job_status}. Message:\n{status.get('message')}"  # noqa: E501
                     raise RuntimeError(
-                        f"RayJob {namespace}/{name} has an unknown status: {job_status}. Message:\n{status.get('message')}"
+                        msg,
                     )
 
             time.sleep(self.poll_interval)
diff --git a/dagster_ray/kuberay/resources.py b/dagster_ray/kuberay/resources.py
index 8062872a..b1a5eb7f 100644
--- a/dagster_ray/kuberay/resources.py
+++ b/dagster_ray/kuberay/resources.py
@@ -1,11 +1,12 @@
+from __future__ import annotations
+
 import contextlib
 import hashlib
 import random
 import re
 import string
 import sys
-from collections.abc import Generator
-from typing import TYPE_CHECKING, Any, Optional, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import dagster._check as check
 from dagster import ConfigurableResource, InitResourceContext
@@ -31,34 +32,39 @@
 from dagster_ray.kuberay.client.base import load_kubeconfig
 
 if TYPE_CHECKING:
+    from collections.abc import Generator
+
     import kubernetes
 
 
 @experimental
 class RayClusterClientResource(ConfigurableResource):
-    kube_context: Optional[str] = None
-    kubeconfig_file: Optional[str] = None
+    kube_context: str | None = None
+    kubeconfig_file: str | None = None
 
     _raycluster_client: RayClusterClient = PrivateAttr()
-    _k8s_api: "kubernetes.client.CustomObjectsApi" = PrivateAttr()
-    _k8s_core_api: "kubernetes.client.CoreV1Api" = PrivateAttr()
+    _k8s_api: kubernetes.client.CustomObjectsApi = PrivateAttr()
+    _k8s_core_api: kubernetes.client.CoreV1Api = PrivateAttr()
 
     @property
     def client(self) -> RayClusterClient:
         if self._raycluster_client is None:
-            raise ValueError(f"{self.__class__.__name__} not initialized")
+            msg = f"{self.__class__.__name__} not initialized"
+            raise ValueError(msg)
         return self._raycluster_client
 
     @property
-    def k8s(self) -> "kubernetes.client.CustomObjectsApi":
+    def k8s(self) -> kubernetes.client.CustomObjectsApi:
         if self._k8s_api is None:
-            raise ValueError(f"{self.__class__.__name__} not initialized")
+            msg = f"{self.__class__.__name__} not initialized"
+            raise ValueError(msg)
         return self._k8s_api
 
     @property
-    def k8s_core(self) -> "kubernetes.client.CoreV1Api":
+    def k8s_core(self) -> kubernetes.client.CoreV1Api:
         if self._k8s_core_api is None:
-            raise ValueError(f"{self.__class__.__name__} not initialized")
+            msg = f"{self.__class__.__name__} not initialized"
+            raise ValueError(msg)
         return self._k8s_core_api
 
     def setup_for_execution(self, context: InitResourceContext) -> None:
@@ -73,9 +79,8 @@ def setup_for_execution(self, context: InitResourceContext) -> None:
 
 @experimental
 class KubeRayCluster(BaseRayResource):
-    """
-    Provides a RayCluster for the current step selection
-    The cluster is automatically deleted after steps execution
+    """Provides a RayCluster for the current step selection
+    The cluster is automatically deleted after steps execution.
     """
 
     deployment_name: str = Field(
@@ -95,7 +100,8 @@ class KubeRayCluster(BaseRayResource):
     @property
     def host(self) -> str:
         if self._host is None:
-            raise ValueError(f"{self.__class__.__name__} not initialized")
+            msg = f"{self.__class__.__name__} not initialized"
+            raise ValueError(msg)
         return self._host
 
     @property
@@ -105,7 +111,8 @@ def namespace(self) -> str:
     @property
     def cluster_name(self) -> str:
         if self._cluster_name is None:
-            raise ValueError(f"{self.__class__.__name__}not initialized")
+            msg = f"{self.__class__.__name__}not initialized"
+            raise ValueError(msg)
         return self._cluster_name
 
     def get_dagster_tags(self, context: InitResourceContext) -> dict[str, str]:
@@ -122,8 +129,6 @@ def yield_for_execution(self, context: InitResourceContext) -> Generator[Self, N
 
         self._cluster_name = self._get_ray_cluster_step_name(context)
 
-        # self._host = f"{self.cluster_name}-head-svc.{self.namespace}.svc.cluster.local"
-
         try:
             # just a safety measure, no need to recreate the cluster for step retries or smth
             if not self.client.client.list(
@@ -137,10 +142,11 @@ def yield_for_execution(self, context: InitResourceContext) -> Generator[Self, N
 
                 resource = self.client.client.create(body=cluster_body, namespace=self.namespace)
                 if not resource:
-                    raise RuntimeError(f"Couldn't create RayCluster {self.namespace}/{self.cluster_name}")
+                    msg = f"Couldn't create RayCluster {self.namespace}/{self.cluster_name}"
+                    raise RuntimeError(msg)
 
                 context.log.info(
-                    f"Created RayCluster {self.namespace}/{self.cluster_name}. Waiting for it to become ready..."
+                    f"Created RayCluster {self.namespace}/{self.cluster_name}. Waiting for it to become ready...",
                 )
 
                 self._wait_raycluster_ready()
@@ -152,7 +158,7 @@ def yield_for_execution(self, context: InitResourceContext) -> Generator[Self, N
                 context.log.info("RayCluster is ready! Connection command:")
 
                 context.log.info(
-                    f"`kubectl -n {self.namespace} port-forward svc/{self.cluster_name}-head-svc 8265:8265 6379:6379 10001:10001`"
+                    f"`kubectl -n {self.namespace} port-forward svc/{self.cluster_name}-head-svc 8265:8265 6379:6379 10001:10001`",  # noqa: E501
                 )
 
             context.log.debug(f"Ray host: {self.host}")
@@ -163,10 +169,10 @@ def yield_for_execution(self, context: InitResourceContext) -> Generator[Self, N
                 self._context = None
 
             yield self
-        except Exception as e:
+        except Exception:
             context.log.critical(f"Couldn't create or connect to RayCluster {self.namespace}/{self.cluster_name}!")
             self._maybe_cleanup_raycluster(context)
-            raise e
+            raise
 
         self._maybe_cleanup_raycluster(context)
         if self._context is not None:
@@ -175,13 +181,12 @@ def yield_for_execution(self, context: InitResourceContext) -> Generator[Self, N
     def _build_raycluster(
         self,
         image: str,
-        labels: Optional[dict[str, str]] = None,  # TODO: use in RayCluster labels
+        labels: dict[str, str] | None = None,  # TODO: use in RayCluster labels
     ) -> dict[str, Any]:
-        """
-        Builds a RayCluster from the provided configuration, while injecting custom image and labels (only known during resource setup)
-        """
+        """Builds a RayCluster from the provided configuration, while injecting custom image
+        and labels (only known during resource setup)."""
         # TODO: inject self.redis_port and self.dashboard_port into the RayCluster configuration
-        # TODO: autoa-apply some tags from dagster-k8s/config
+        # TODO: auto-apply some tags from dagster-k8s/config
 
         labels = labels or {}
         assert isinstance(labels, dict)
@@ -190,18 +195,17 @@ def _build_raycluster(
         head_group_spec = self.ray_cluster.head_group_spec.copy()
         worker_group_specs = self.ray_cluster.worker_group_specs.copy()
 
-        def update_group_spec(group_spec: dict[str, Any]):
+        def update_group_spec(group_spec: dict[str, Any]) -> None:
             # TODO: only inject if the container has a `dagster.io/inject-image` annotation or smth
             if group_spec["template"]["spec"]["containers"][0].get("image") is None:
                 group_spec["template"]["spec"]["containers"][0]["image"] = image
 
             if group_spec.get("metadata") is None:
                 group_spec["metadata"] = {"labels": labels}
+            elif group_spec["metadata"].get("labels") is None:
+                group_spec["metadata"]["labels"] = labels
             else:
-                if group_spec["metadata"].get("labels") is None:
-                    group_spec["metadata"]["labels"] = labels
-                else:
-                    group_spec["metadata"]["labels"].update(labels)
+                group_spec["metadata"]["labels"].update(labels)
 
         update_group_spec(head_group_spec)
         for worker_group_spec in worker_group_specs:
@@ -222,7 +226,7 @@ def update_group_spec(group_spec: dict[str, Any]):
             },
         }
 
-    def _wait_raycluster_ready(self):
+    def _wait_raycluster_ready(self) -> None:
         import kubernetes
 
         self.client.client.wait_until_ready(self.cluster_name, namespace=self.namespace, timeout=self.timeout)
@@ -246,7 +250,7 @@ def _wait_raycluster_ready(self):
                 w.stop()
                 return
 
-    def _maybe_cleanup_raycluster(self, context: InitResourceContext):
+    def _maybe_cleanup_raycluster(self, context: InitResourceContext) -> None:
         assert context.log is not None
 
         if not self.skip_cleanup and cast(DagsterRun, context.dagster_run).status != DagsterRunStatus.FAILURE:
@@ -256,7 +260,7 @@ def _maybe_cleanup_raycluster(self, context: InitResourceContext):
             context.log.warning(
                 f"Skipping RayCluster {self.cluster_name} deletion because `disable_cluster_cleanup` "
                 f"config parameter is set to `True` or the run failed. "
-                f"It may be still be deleted by the automatic cleanup job."
+                f"It may be still be deleted by the automatic cleanup job.",
             )
 
     def _get_ray_cluster_step_name(self, context: InitResourceContext) -> str:
@@ -278,12 +282,10 @@ def _get_ray_cluster_step_name(self, context: InitResourceContext) -> str:
         )
 
         step_name = f"{cluster_name_prefix}-{name_key}".lower()
-        step_name = re.sub(r"[^-0-9a-z]", "-", step_name)
-
-        return step_name
+        return re.sub(r"[^-0-9a-z]", "-", step_name)
 
 
-def get_k8s_object_name(run_id: str, step_key: Optional[str] = None):
+def get_k8s_object_name(run_id: str, step_key: str | None = None):
     """Creates a unique (short!) identifier to name k8s objects based on run ID and step key(s).
 
     K8s Job names are limited to 63 characters, because they are used as labels. For more info, see:
diff --git a/dagster_ray/pipes.py b/dagster_ray/pipes.py
index 68af3f4f..a19a847a 100644
--- a/dagster_ray/pipes.py
+++ b/dagster_ray/pipes.py
@@ -1,10 +1,11 @@
+from __future__ import annotations
+
 import random
 import string
 import threading
 import time
-from collections.abc import Generator, Iterator
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union, cast
+from typing import TYPE_CHECKING, Any, TypedDict, Union, cast
 
 import dagster._check as check
 from dagster import AssetExecutionContext, OpExecutionContext, PipesClient
@@ -16,7 +17,6 @@
     PipesContextInjector,
     PipesMessageReader,
 )
-from dagster._core.pipes.context import PipesMessageHandler, PipesSession
 from dagster._core.pipes.utils import (
     PipesEnvContextInjector,
     _join_thread,
@@ -29,6 +29,9 @@
 from dagster_ray._base.utils import get_dagster_tags
 
 if TYPE_CHECKING:
+    from collections.abc import Generator, Iterator
+
+    from dagster._core.pipes.context import PipesMessageHandler, PipesSession
     from ray.job_submission import JobStatus, JobSubmissionClient
 
 
@@ -37,18 +40,17 @@
 
 @experimental
 class PipesRayJobMessageReader(PipesMessageReader):
-    """
-    Dagster Pipes message reader for receiving messages from a Ray job.
+    """Dagster Pipes message reader for receiving messages from a Ray job.
     Will extract Dagster events and forward the rest to stdout.
     """
 
-    def __init__(self):
-        self._handler: Optional[PipesMessageHandler] = None
-        self._thread: Optional[threading.Thread] = None
+    def __init__(self) -> None:
+        self._handler: PipesMessageHandler | None = None
+        self._thread: threading.Thread | None = None
         self.session_closed = threading.Event()
 
     @contextmanager
-    def read_messages(self, handler: "PipesMessageHandler") -> Iterator[PipesParams]:
+    def read_messages(self, handler: PipesMessageHandler) -> Iterator[PipesParams]:
         #  This method should start a thread to continuously read messages from some location
         self._handler = handler
 
@@ -58,9 +60,10 @@ def read_messages(self, handler: "PipesMessageHandler") -> Iterator[PipesParams]
             self.terminate()
 
     @property
-    def handler(self) -> "PipesMessageHandler":
+    def handler(self) -> PipesMessageHandler:
         if self._handler is None:
-            raise Exception("PipesMessageHandler is only available while reading messages in open_pipes_session")
+            msg = "PipesMessageHandler is only available while reading messages in open_pipes_session"
+            raise Exception(msg)
 
         return self._handler
 
@@ -73,7 +76,7 @@ def terminate(self) -> None:
         self._handler = None
 
     # TODO: call this method as part of self.read_messages
-    def consume_job_logs(self, client: "JobSubmissionClient", job_id: str, blocking: bool = False) -> None:
+    def consume_job_logs(self, client: JobSubmissionClient, job_id: str, blocking: bool = False) -> None:
         if blocking:
             handle_job_logs(handler=self.handler, client=client, job_id=job_id, session_closed=None)
         else:
@@ -94,8 +97,11 @@ def no_messages_debug_text(self) -> str:
 
 
 def handle_job_logs(
-    handler: PipesMessageHandler, client: "JobSubmissionClient", job_id: str, session_closed: Optional[threading.Event]
-):
+    handler: PipesMessageHandler,
+    client: JobSubmissionClient,
+    job_id: str,
+    session_closed: threading.Event | None,
+) -> None:
     import asyncio
 
     async_tailer = client.tail_job_logs(job_id=job_id)
@@ -174,21 +180,23 @@ class PipesRayJobClient(PipesClient, TreatAsResourceParam):
             context into the Ray job. Defaults to :py:class:`PipesEnvContextInjector`.
         message_reader (Optional[PipesMessageReader]): A message reader to use to read messages
             from the glue job run. Defaults to :py:class:`PipesRayJobMessageReader`.
-        forward_termination (bool): Whether to cancel the `RayJob` job run when the Dagster process receives a termination signal.
+        forward_termination (bool): Whether to cancel the `RayJob` job run when the Dagster process receives
+            a termination signal.
         timeout (int): Timeout for various internal interactions with the Kubernetes RayJob.
         poll_interval (int): Interval at which to poll the Kubernetes for status updates.
         Is useful when running in a local environment.
+
     """
 
     def __init__(
         self,
-        client: "JobSubmissionClient",
-        context_injector: Optional[PipesContextInjector] = None,
-        message_reader: Optional[PipesMessageReader] = None,
+        client: JobSubmissionClient,
+        context_injector: PipesContextInjector | None = None,
+        message_reader: PipesMessageReader | None = None,
         forward_termination: bool = True,
         timeout: int = 600,
         poll_interval: int = 5,
-    ):
+    ) -> None:
         self.client = client
         self._context_injector = context_injector or PipesEnvContextInjector()
         self._message_reader = message_reader or PipesRayJobMessageReader()
@@ -197,24 +205,23 @@ def __init__(
         self.timeout = check.int_param(timeout, "timeout")
         self.poll_interval = check.int_param(poll_interval, "poll_interval")
 
-        self._job_submission_client: Optional[JobSubmissionClient] = None
+        self._job_submission_client: JobSubmissionClient | None = None
 
     def run(  # type: ignore
         self,
         *,
         context: OpOrAssetExecutionContext,
         submit_job_params: SubmitJobParams,
-        extras: Optional[PipesExtras] = None,
+        extras: PipesExtras | None = None,
     ) -> PipesClientCompletedInvocation:
-        """
-        Execute a RayJob, enriched with the Pipes protocol.
+        """Execute a RayJob, enriched with the Pipes protocol.
 
         Args:
             context (OpExecutionContext): Current Dagster op or asset context.
             ray_job (Dict[str, Any]): RayJob specification. `API reference <https://ray-project.github.io/kuberay/reference/api/#rayjob>`_.
             extras (Optional[Dict[str, Any]]): Additional information to pass to the Pipes session.
-        """
 
+        """
         with open_pipes_session(
             context=context,
             message_reader=self._message_reader,
@@ -237,11 +244,13 @@ def run(  # type: ignore
                 raise
 
     def get_dagster_tags(self, context: OpOrAssetExecutionContext) -> dict[str, str]:
-        tags = get_dagster_tags(context)
-        return tags
+        return get_dagster_tags(context)
 
     def _enrich_submit_job_params(
-        self, context: OpOrAssetExecutionContext, session: PipesSession, submit_job_params: SubmitJobParams
+        self,
+        context: OpOrAssetExecutionContext,
+        session: PipesSession,
+        submit_job_params: SubmitJobParams,
     ) -> EnrichedSubmitJobParams:
         runtime_env = submit_job_params.get("runtime_env", {})
         metadata = submit_job_params.get("metadata", {})
@@ -279,7 +288,7 @@ def _read_messages(self, context: OpOrAssetExecutionContext, job_id: str) -> Non
                 blocking=True,
             )
 
-    def _wait_for_completion(self, context: OpOrAssetExecutionContext, job_id: str) -> "JobStatus":
+    def _wait_for_completion(self, context: OpOrAssetExecutionContext, job_id: str) -> JobStatus:
         from ray.job_submission import JobStatus
 
         context.log.info(f"[pipes] Waiting for RayJob {job_id} to complete...")
@@ -290,8 +299,9 @@ def _wait_for_completion(self, context: OpOrAssetExecutionContext, job_id: str)
             if status.is_terminal():
                 if status in {JobStatus.FAILED, JobStatus.STOPPED}:
                     job_details = self.client.get_job_info(job_id)
+                    msg = f"[pipes] RayJob {job_id} failed with status {status}. Message:\n{job_details.message}"
                     raise RuntimeError(
-                        f"[pipes] RayJob {job_id} failed with status {status}. Message:\n{job_details.message}"
+                        msg,
                     )
 
                 return status
diff --git a/dagster_ray/resources.py b/dagster_ray/resources.py
index e05fff49..650501d8 100644
--- a/dagster_ray/resources.py
+++ b/dagster_ray/resources.py
@@ -19,8 +19,7 @@
 
 
 class LocalRay(BaseRayResource):
-    """
-    Dummy Resource.
+    """Dummy Resource.
     Is useful for testing and local development.
     Provides the same interface as actual Resources.
     """
diff --git a/dagster_ray/run_launcher.py b/dagster_ray/run_launcher.py
index 0c32f115..b5314f7d 100644
--- a/dagster_ray/run_launcher.py
+++ b/dagster_ray/run_launcher.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 import sys
 from typing import TYPE_CHECKING, Any, Optional, cast
@@ -5,7 +7,6 @@
 import dagster
 from dagster import _check as check
 from dagster._cli.api import ExecuteRunArgs  # type: ignore
-from dagster._config.config_schema import UserConfigSchema
 from dagster._core.events import EngineEventData
 from dagster._core.launcher import LaunchRunContext, ResumeRunContext, RunLauncher
 from dagster._core.launcher.base import CheckRunHealthResult, WorkerStatus
@@ -21,17 +22,20 @@
 from dagster_ray.utils import resolve_env_vars_list
 
 if TYPE_CHECKING:
+    from dagster._config.config_schema import UserConfigSchema
     from ray.job_submission import JobSubmissionClient
 
 
-def get_job_submission_id_from_run_id(run_id: str, resume_attempt_number=None):
+def get_job_submission_id_from_run_id(run_id: str, resume_attempt_number=None) -> str:
     return f"dagster-run-{run_id}" + ("" if not resume_attempt_number else f"-{resume_attempt_number}")
 
 
 class RayLauncherConfig(RayExecutionConfig, RayJobSubmissionClientConfig):
-    env_vars: Optional[list[str]] = Field(
+    env_vars: list[str] | None = Field(
         default=None,
-        description="A list of environment variables to inject into the Job. Each can be of the form KEY=VALUE or just KEY (in which case the value will be pulled from the current process).",
+        description="""A list of environment variables to inject into the Job.
+        Each can be of the form KEY=VALUE or just KEY (in which case the value will
+        be pulled from the current process).""",
     )
 
 
@@ -39,17 +43,17 @@ class RayRunLauncher(RunLauncher, ConfigurableClass):
     def __init__(
         self,
         address: str,
-        metadata: Optional[dict[str, Any]] = None,
-        headers: Optional[dict[str, Any]] = None,
-        cookies: Optional[dict[str, Any]] = None,
-        env_vars: Optional[list[str]] = None,
-        runtime_env: Optional[dict[str, Any]] = None,
-        num_cpus: Optional[int] = None,
-        num_gpus: Optional[int] = None,
-        memory: Optional[int] = None,
-        resources: Optional[dict[str, float]] = None,
-        inst_data: Optional[ConfigurableClassData] = None,
-    ):
+        metadata: dict[str, Any] | None = None,
+        headers: dict[str, Any] | None = None,
+        cookies: dict[str, Any] | None = None,
+        env_vars: list[str] | None = None,
+        runtime_env: dict[str, Any] | None = None,
+        num_cpus: int | None = None,
+        num_gpus: int | None = None,
+        memory: int | None = None,
+        resources: dict[str, float] | None = None,
+        inst_data: ConfigurableClassData | None = None,
+    ) -> None:
         """RunLauncher that starts a Ray job (incluster mode) for each Dagster run.
 
         Encapsulates each run in a separate, isolated invocation of ``ray.job_submission.JobSubmissionClient``.
@@ -84,13 +88,13 @@ def __init__(
         super().__init__()
 
     @property
-    def client(self) -> "JobSubmissionClient":  # note: this must be a property
+    def client(self) -> JobSubmissionClient:  # note: this must be a property
         from ray.job_submission import JobSubmissionClient
 
         return JobSubmissionClient(self.address, metadata=self.metadata, headers=self.headers, cookies=self.cookies)
 
     @property
-    def inst_data(self) -> Optional[ConfigurableClassData]:
+    def inst_data(self) -> ConfigurableClassData | None:
         return self._inst_data
 
     @classmethod
@@ -102,15 +106,15 @@ def from_config_value(cls, inst_data, config_value):
         return cls(inst_data=inst_data, **config_value["ray"])
 
     @property
-    def supports_resume_run(self):
+    def supports_resume_run(self) -> bool:
         return True
 
     @property
-    def supports_check_run_worker_health(self):
+    def supports_check_run_worker_health(self) -> bool:
         return True
 
     @property
-    def supports_run_worker_crash_recovery(self):
+    def supports_run_worker_crash_recovery(self) -> bool:
         return True
 
     def launch_run(self, context: LaunchRunContext) -> None:
@@ -124,7 +128,7 @@ def launch_run(self, context: LaunchRunContext) -> None:
                 run_id=run.run_id,
                 instance_ref=self._instance.get_ref(),
                 set_exit_code_on_failure=True,
-            ).get_command_args()
+            ).get_command_args(),
         )
 
         # wrap the json in quotes to prevent erros with shell commands
@@ -132,7 +136,7 @@ def launch_run(self, context: LaunchRunContext) -> None:
 
         self._launch_ray_job(submission_id, " ".join(args), run)
 
-    def _launch_ray_job(self, submission_id: str, entrypoint: str, run: DagsterRun):
+    def _launch_ray_job(self, submission_id: str, entrypoint: str, run: DagsterRun) -> None:
         # note: entrypoint is a shell command
         job_origin = check.not_none(run.job_code_origin)
 
@@ -166,7 +170,7 @@ def _launch_ray_job(self, submission_id: str, entrypoint: str, run: DagsterRun):
         runtime_env["env_vars"].update(
             {
                 "DAGSTER_RUN_JOB_NAME": job_origin.job_name,
-            }
+            },
         )
 
         self._instance.report_engine_event(
@@ -176,7 +180,7 @@ def _launch_ray_job(self, submission_id: str, entrypoint: str, run: DagsterRun):
                 {
                     "Ray Job Submission ID": submission_id,
                     "Run ID": run.run_id,
-                }
+                },
             ),
             cls=self.__class__,
         )
@@ -201,7 +205,8 @@ def _launch_ray_job(self, submission_id: str, entrypoint: str, run: DagsterRun):
     def resume_run(self, context: ResumeRunContext) -> None:
         run = context.dagster_run
         submission_id = get_job_submission_id_from_run_id(
-            run.run_id, resume_attempt_number=context.resume_attempt_number
+            run.run_id,
+            resume_attempt_number=context.resume_attempt_number,
         )
         job_origin = check.not_none(run.job_code_origin)
 
@@ -211,7 +216,7 @@ def resume_run(self, context: ResumeRunContext) -> None:
                 run_id=run.run_id,
                 instance_ref=self._instance.get_ref(),
                 set_exit_code_on_failure=True,
-            ).get_command_args()
+            ).get_command_args(),
         )
 
         # wrap the json in quotes to prevent erros with shell commands
@@ -229,7 +234,8 @@ def terminate(self, run_id: str) -> bool:
         self._instance.report_run_canceling(run)
 
         submission_id = get_job_submission_id_from_run_id(
-            run.run_id, resume_attempt_number=self._instance.count_resume_run_attempts(run.run_id)
+            run.run_id,
+            resume_attempt_number=self._instance.count_resume_run_attempts(run.run_id),
         )
 
         try:
@@ -257,8 +263,10 @@ def terminate(self, run_id: str) -> bool:
             return False
 
     def get_run_worker_debug_info(
-        self, run: DagsterRun, include_container_logs: Optional[bool] = True
-    ) -> Optional[str]:
+        self,
+        run: DagsterRun,
+        include_container_logs: bool | None = True,
+    ) -> str | None:
         try:
             job_details = [j for j in self.client.list_jobs() if (j.metadata or {}).get("dagster/run-id") == run.run_id]
 
@@ -281,27 +289,30 @@ def check_run_worker_health(self, run: DagsterRun):
             status = self.client.get_job_status(submission_id)
         except RuntimeError:
             return CheckRunHealthResult(
-                WorkerStatus.UNKNOWN, str(serializable_error_info_from_exc_info(sys.exc_info()))
+                WorkerStatus.UNKNOWN,
+                str(serializable_error_info_from_exc_info(sys.exc_info())),
             )
 
         # If the run is in a non-terminal (and non-STARTING) state but the ray job is not active,
         # something went wrong
         if run.status in (DagsterRunStatus.STARTED, DagsterRunStatus.CANCELING) and status.is_terminal():
             return CheckRunHealthResult(
-                WorkerStatus.FAILED, f"Run has not completed but Ray job has is in status: {status}"
+                WorkerStatus.FAILED,
+                f"Run has not completed but Ray job has is in status: {status}",
             )
 
-        elif status == JobStatus.FAILED:
+        if status == JobStatus.FAILED:
             job_details = self.client.get_job_info(submission_id)
             return CheckRunHealthResult(WorkerStatus.FAILED, f"Ray job failed. Message: {job_details.message}")
-        elif status == JobStatus.STOPPED:
+        if status == JobStatus.STOPPED:
             job_details = self.client.get_job_info(submission_id)
             return CheckRunHealthResult(
-                WorkerStatus.FAILED, f"Ray job has been stopped externally. Message: {job_details.message}"
+                WorkerStatus.FAILED,
+                f"Ray job has been stopped externally. Message: {job_details.message}",
             )
-        elif status == JobStatus.SUCCEEDED:
+        if status == JobStatus.SUCCEEDED:
             return CheckRunHealthResult(WorkerStatus.SUCCESS)
-        elif status in {JobStatus.RUNNING, JobStatus.PENDING}:
+        if status in {JobStatus.RUNNING, JobStatus.PENDING}:
             return CheckRunHealthResult(WorkerStatus.RUNNING)
 
         # safe return in case more statuses are introduced later
diff --git a/dagster_ray/utils.py b/dagster_ray/utils.py
index ef7b2076..9c4bdb72 100644
--- a/dagster_ray/utils.py
+++ b/dagster_ray/utils.py
@@ -1,8 +1,9 @@
+from __future__ import annotations
+
 import os
-from typing import Optional
 
 
-def resolve_env_vars_list(env_vars: Optional[list[str]]) -> dict[str, str]:
+def resolve_env_vars_list(env_vars: list[str] | None) -> dict[str, str]:
     res = {}
 
     if env_vars is not None:
@@ -10,8 +11,7 @@ def resolve_env_vars_list(env_vars: Optional[list[str]]) -> dict[str, str]:
             if "=" in env_var:
                 var, value = env_var.split("=", 1)
                 res[var] = value
-            else:
-                if value := os.getenv(env_var):
-                    res[env_var] = value
+            elif value := os.getenv(env_var):
+                res[env_var] = value
 
     return res
diff --git a/examples/docker/run_launcher/definitions.py b/examples/docker/run_launcher/definitions.py
index 22b5d3da..b0082b81 100644
--- a/examples/docker/run_launcher/definitions.py
+++ b/examples/docker/run_launcher/definitions.py
@@ -41,7 +41,7 @@ def sum_one_and_two(a: int, b: int) -> int:
 
 
 @job(tags={"dagster-ray/config": {"num_cpus": 0.5}})
-def my_job():
+def my_job() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
     sum_one_and_two(return_one_result, return_two_result)
diff --git a/examples/local/executor/definitions.py b/examples/local/executor/definitions.py
index a4668248..d8853169 100644
--- a/examples/local/executor/definitions.py
+++ b/examples/local/executor/definitions.py
@@ -36,7 +36,7 @@ def sum_one_and_two(a: int, b: int) -> int:
 
 
 @job(executor_def=ray_executor)
-def my_job():
+def my_job() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
     sum_one_and_two(return_one_result, return_two_result)
diff --git a/examples/local/run_launcher/definitions.py b/examples/local/run_launcher/definitions.py
index bd1791b3..82f479da 100644
--- a/examples/local/run_launcher/definitions.py
+++ b/examples/local/run_launcher/definitions.py
@@ -34,7 +34,7 @@ def sum_one_and_two(a: int, b: int) -> int:
 
 
 @job(tags={"dagster-ray/config": {"num_cpus": 0.5}})
-def my_job():
+def my_job() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
 
diff --git a/examples/local/run_launcher_and_executor/definitions.py b/examples/local/run_launcher_and_executor/definitions.py
index a4668248..d8853169 100644
--- a/examples/local/run_launcher_and_executor/definitions.py
+++ b/examples/local/run_launcher_and_executor/definitions.py
@@ -36,7 +36,7 @@ def sum_one_and_two(a: int, b: int) -> int:
 
 
 @job(executor_def=ray_executor)
-def my_job():
+def my_job() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
     sum_one_and_two(return_one_result, return_two_result)
diff --git a/pyproject.toml b/pyproject.toml
index 23644394..b43ce203 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -109,7 +109,21 @@ exclude = [
     "venv",
 ]
 [tool.ruff.lint]
-extend-select = ["I", "TID252", "TID253", "UP"]
+select = ["ALL"]
+ignore = [
+    # disallow assert statements
+    "S101",
+    # disallow non-specific type: ignore comments
+    "PGH003",
+    # docs
+    "D",
+    # missing trailing comma
+    "COM812",
+    # fixme and TODO comments
+    "FIX", "TD",
+    # logging call with f-strings (doesn't make sense since it's almost always the Dagster logger being used)
+    "G004",
+]
 [tool.ruff.lint.isort]
 known-first-party = ["dagster_ray", "tests"]
 
diff --git a/scripts/check_deps.py b/scripts/check_deps.py
index bde53726..a8f4263e 100644
--- a/scripts/check_deps.py
+++ b/scripts/check_deps.py
@@ -13,12 +13,13 @@ def check_python_312(
     python: Version,
     dagster: Version,
     ray: Version,
-):
+) -> None:
     if python >= Version("3.12") and ray < Version("2.37.0"):
-        raise RuntimeError("Ray version must be >=2.37.0 for Python >=3.12")
+        msg = "Ray version must be >=2.37.0 for Python >=3.12"
+        raise RuntimeError(msg)
 
 
-def main():
+def main() -> None:
     args = parser.parse_args()
     check_python_312(
         python=args.python,
diff --git a/tests/conftest.py b/tests/conftest.py
index fc86b655..ab530383 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -18,7 +18,8 @@ def local_ray_address() -> Iterator[str]:
     import ray
 
     context = ray.init(
-        ignore_reinit_error=True, runtime_env={"env_vars": {"RAY_ENABLE_RECORD_ACTOR_TASK_LOGGING": "1"}}
+        ignore_reinit_error=True,
+        runtime_env={"env_vars": {"RAY_ENABLE_RECORD_ACTOR_TASK_LOGGING": "1"}},
     )
 
     yield "auto"
diff --git a/tests/examples/test_examples.py b/tests/examples/test_examples.py
index f9454cba..23fa3e16 100644
--- a/tests/examples/test_examples.py
+++ b/tests/examples/test_examples.py
@@ -15,9 +15,10 @@
 
 
 @pytest.mark.parametrize(
-    "example_dir", [RUN_LAUNCHER_EXAMPLE_DIR, EXECUTOR_EXAMPLE_DIR, RUN_LAUNCHER_AND_EXECUTOR_EXAMPLE_DIR]
+    "example_dir",
+    [RUN_LAUNCHER_EXAMPLE_DIR, EXECUTOR_EXAMPLE_DIR, RUN_LAUNCHER_AND_EXECUTOR_EXAMPLE_DIR],
 )
-def test_ray_run_launcher(local_ray_address: str, example_dir: Path, tmp_path_factory):
+def test_ray_run_launcher(local_ray_address: str, example_dir: Path, tmp_path_factory) -> None:
     dagster_home = tmp_path_factory.mktemp("dagster_home")
 
     # copy dagter.yaml from example_dir to dagster_home
diff --git a/tests/kuberay/conftest.py b/tests/kuberay/conftest.py
index 2b048880..b407a30f 100644
--- a/tests/kuberay/conftest.py
+++ b/tests/kuberay/conftest.py
@@ -20,7 +20,7 @@
 
 
 @pytest.fixture(scope="session")
-def kuberay_helm_repo():
+def kuberay_helm_repo() -> None:
     subprocess.run(["helm", "repo", "add", "kuberay", "https://ray-project.github.io/kuberay-helm/"], check=True)
     subprocess.run(["helm", "repo", "update", "kuberay"], check=True)
 
@@ -90,7 +90,10 @@ def dagster_ray_image():
 @pytest_cases.fixture(scope="session")  # type: ignore
 @pytest.mark.parametrize("kuberay_version", KUBERAY_VERSIONS)
 def k8s_with_kuberay(
-    request, kuberay_helm_repo, dagster_ray_image: str, kuberay_version: str
+    request,
+    kuberay_helm_repo,
+    dagster_ray_image: str,
+    kuberay_version: str,
 ) -> Iterator[AClusterManager]:
     k8s = select_provider_manager("minikube")("dagster-ray")
     k8s.create(ClusterOptions(api_version=KUBERNETES_VERSION))
diff --git a/tests/kuberay/test_pipes.py b/tests/kuberay/test_pipes.py
index 030b12ec..c11d9ecc 100644
--- a/tests/kuberay/test_pipes.py
+++ b/tests/kuberay/test_pipes.py
@@ -1,5 +1,3 @@
-import sys
-
 import pytest
 import ray  # noqa: TID253
 from dagster import AssetExecutionContext, DagsterEventType, EventRecordsFilter, asset, materialize
@@ -46,7 +44,7 @@
                                 "imagePullPolicy": "IfNotPresent",
                                 "name": "ray-head",
                                 "securityContext": {"runAsUser": 0},
-                            }
+                            },
                         ],
                         # "serviceAccountName": "ray",
                         "volumes": [],
@@ -73,17 +71,15 @@ def pipes_kube_rayjob_client(k8s_with_kuberay: AClusterManager):
     )
 
 
-def test_rayjob_pipes(pipes_kube_rayjob_client: PipesKubeRayJobClient, dagster_ray_image: str, capsys):
+def test_rayjob_pipes(pipes_kube_rayjob_client: PipesKubeRayJobClient, dagster_ray_image: str, capsys) -> None:
     @asset
     def my_asset(context: AssetExecutionContext, pipes_kube_rayjob_client: PipesKubeRayJobClient):
-        result = pipes_kube_rayjob_client.run(
+        return pipes_kube_rayjob_client.run(
             context=context,
             ray_job=RAY_JOB,
             extras={"foo": "bar"},
         ).get_materialize_result()
 
-        return result
-
     with instance_for_test() as instance:
         result = materialize(
             [my_asset],
@@ -94,9 +90,6 @@ def my_asset(context: AssetExecutionContext, pipes_kube_rayjob_client: PipesKube
 
         captured = capsys.readouterr()
 
-        print(captured.out)
-        print(captured.err, file=sys.stderr)
-
         mat_evts = result.get_asset_materialization_events()
 
         mat = instance.get_latest_materialization_event(my_asset.key)
@@ -106,7 +99,8 @@ def my_asset(context: AssetExecutionContext, pipes_kube_rayjob_client: PipesKube
 
         assert result.success
         assert mat
-        assert mat and mat.asset_materialization
+        assert mat
+        assert mat.asset_materialization
         assert mat.asset_materialization.metadata["some_metric"].value == 0
         assert mat.asset_materialization.tags
         assert mat.asset_materialization.tags[DATA_VERSION_TAG] == "alpha"
@@ -123,21 +117,19 @@ def pipes_ray_job_client(k8s_with_raycluster: tuple[dict[str, str], AClusterMana
     return PipesRayJobClient(
         client=JobSubmissionClient(
             address=hosts["dashboard"],
-        )
+        ),
     )
 
 
-def test_ray_job_pipes(pipes_ray_job_client: PipesRayJobClient, capsys):
+def test_ray_job_pipes(pipes_ray_job_client: PipesRayJobClient, capsys) -> None:
     @asset
     def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobClient):
-        result = pipes_ray_job_client.run(
+        return pipes_ray_job_client.run(
             context=context,
             submit_job_params={"entrypoint": ENTRYPOINT, "entrypoint_num_cpus": 0.1},
             extras={"foo": "bar"},
         ).get_materialize_result()
 
-        return result
-
     with instance_for_test() as instance:
         result = materialize(
             [my_asset],
@@ -147,9 +139,6 @@ def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobCl
 
         captured = capsys.readouterr()
 
-        print(captured.out)
-        print(captured.err, file=sys.stderr)
-
         mat_evts = result.get_asset_materialization_events()
 
         mat = instance.get_latest_materialization_event(my_asset.key)
@@ -159,7 +148,8 @@ def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobCl
 
         assert result.success
         assert mat
-        assert mat and mat.asset_materialization
+        assert mat
+        assert mat.asset_materialization
         assert mat.asset_materialization.metadata["some_metric"].value == 0
         assert mat.asset_materialization.tags
         assert mat.asset_materialization.tags[DATA_VERSION_TAG] == "alpha"
diff --git a/tests/kuberay/test_raycluster.py b/tests/kuberay/test_raycluster.py
index 9bcbbc6c..457f2140 100644
--- a/tests/kuberay/test_raycluster.py
+++ b/tests/kuberay/test_raycluster.py
@@ -70,7 +70,7 @@ def get_hostname():
 def test_kuberay_cluster_resource(
     ray_cluster_resource: KubeRayCluster,
     k8s_with_kuberay: AClusterManager,
-):
+) -> None:
     @asset
     # testing RayResource type annotation too!
     def my_asset(context: AssetExecutionContext, ray_cluster: RayResource) -> None:
@@ -96,7 +96,8 @@ def my_asset(context: AssetExecutionContext, ray_cluster: RayResource) -> None:
             assert ray_cluster.cluster_name in ray.get(get_hostname.remote())
 
             ray_cluster_description = ray_cluster.client.client.get(
-                ray_cluster.cluster_name, namespace=ray_cluster.namespace
+                ray_cluster.cluster_name,
+                namespace=ray_cluster.namespace,
             )
             assert ray_cluster_description["metadata"]["labels"]["dagster.io/run_id"] == context.run_id
             assert ray_cluster_description["metadata"]["labels"]["dagster.io/cluster"] == ray_cluster.cluster_name
@@ -113,8 +114,9 @@ def my_asset(context: AssetExecutionContext, ray_cluster: RayResource) -> None:
     assert (
         len(
             kuberay_client.list(
-                namespace=ray_cluster_resource.namespace, label_selector=f"dagster.io/run_id={result.run_id}"
-            )["items"]
+                namespace=ray_cluster_resource.namespace,
+                label_selector=f"dagster.io/run_id={result.run_id}",
+            )["items"],
         )
         == 0
     )
@@ -123,7 +125,7 @@ def my_asset(context: AssetExecutionContext, ray_cluster: RayResource) -> None:
 def test_kuberay_cleanup_job(
     ray_cluster_resource_skip_cleanup: KubeRayCluster,
     k8s_with_kuberay: AClusterManager,
-):
+) -> None:
     @asset
     def my_asset(ray_cluster: RayResource) -> None:
         assert isinstance(ray_cluster, KubeRayCluster)
@@ -140,7 +142,7 @@ def my_asset(ray_cluster: RayResource) -> None:
             kuberay_client.list(
                 namespace=ray_cluster_resource_skip_cleanup.namespace,
                 label_selector=f"dagster.io/run_id={result.run_id}",
-            )["items"]
+            )["items"],
         )
         > 0
     )
@@ -153,11 +155,12 @@ def my_asset(ray_cluster: RayResource) -> None:
             ops={
                 "cleanup_kuberay_clusters": CleanupKuberayClustersConfig(
                     namespace=ray_cluster_resource_skip_cleanup.namespace,
-                )
-            }
+                ),
+            },
         ),
     )
 
     assert not kuberay_client.list(
-        namespace=ray_cluster_resource_skip_cleanup.namespace, label_selector=f"dagster.io/run_id={result.run_id}"
+        namespace=ray_cluster_resource_skip_cleanup.namespace,
+        label_selector=f"dagster.io/run_id={result.run_id}",
     )["items"]
diff --git a/tests/scripts/remote_job.py b/tests/scripts/remote_job.py
index 4640cadf..c3ceab35 100644
--- a/tests/scripts/remote_job.py
+++ b/tests/scripts/remote_job.py
@@ -1,5 +1,3 @@
-import sys
-
 from dagster_pipes import open_dagster_pipes
 
 with open_dagster_pipes() as pipes:
@@ -9,5 +7,3 @@
         metadata={"some_metric": {"raw_value": 0, "type": "int"}},
         data_version="alpha",
     )
-    print("Hello from stdout!")
-    print("Hello from stderr!", file=sys.stderr)
diff --git a/tests/test_executor.py b/tests/test_executor.py
index 50e0370a..51a3d4f7 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -34,7 +34,7 @@ def sum_one_and_two(a: int, b: int) -> int:
 
 
 @job(executor_def=ray_executor)
-def my_job():
+def my_job() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
     sum_one_and_two(return_one_result, return_two_result)
@@ -42,11 +42,12 @@ def my_job():
 
 @op
 def failed_op() -> None:
-    raise RuntimeError("This op failed!")
+    msg = "This op failed!"
+    raise RuntimeError(msg)
 
 
 @job(executor_def=ray_executor)
-def my_failing_job():
+def my_failing_job() -> None:
     failed_op()
 
 
@@ -56,7 +57,7 @@ def dagster_instance() -> Iterator[DagsterInstance]:
         yield instance
 
 
-def test_ray_executor(local_ray_address: str, dagster_instance: DagsterInstance):
+def test_ray_executor(local_ray_address: str, dagster_instance: DagsterInstance) -> None:
     result = execute_job(
         job=reconstructable(my_job),
         instance=dagster_instance,
@@ -64,8 +65,8 @@ def test_ray_executor(local_ray_address: str, dagster_instance: DagsterInstance)
             "execution": {
                 "config": {
                     "ray": {"address": local_ray_address},
-                }
-            }
+                },
+            },
         },
     )
 
@@ -76,13 +77,13 @@ def test_ray_executor(local_ray_address: str, dagster_instance: DagsterInstance)
 
 
 @job(executor_def=ray_executor, resource_defs={"io_manager": ray_io_manager})
-def my_job_with_ray_io_manager():
+def my_job_with_ray_io_manager() -> None:
     return_two_result = return_two()
     return_one_result = return_one()
     sum_one_and_two(return_one_result, return_two_result)
 
 
-def test_ray_executor_with_ray_io_manager(local_ray_address: str, dagster_instance: DagsterInstance):
+def test_ray_executor_with_ray_io_manager(local_ray_address: str, dagster_instance: DagsterInstance) -> None:
     result = execute_job(
         job=reconstructable(my_job_with_ray_io_manager),
         instance=dagster_instance,
@@ -90,15 +91,15 @@ def test_ray_executor_with_ray_io_manager(local_ray_address: str, dagster_instan
             "execution": {
                 "config": {
                     "ray": {"address": local_ray_address},
-                }
-            }
+                },
+            },
         },
     )
 
     assert result.success, result.get_step_failure_events()[0].event_specific_data
 
 
-def test_ray_executor_local_failing(local_ray_address: str, dagster_instance: DagsterInstance):
+def test_ray_executor_local_failing(local_ray_address: str, dagster_instance: DagsterInstance) -> None:
     result = execute_job(
         job=reconstructable(my_failing_job),
         instance=dagster_instance,
@@ -106,8 +107,8 @@ def test_ray_executor_local_failing(local_ray_address: str, dagster_instance: Da
             "execution": {
                 "config": {
                     "ray": {"address": local_ray_address},
-                }
-            }
+                },
+            },
         },
     )
     assert result.get_failed_step_keys() == {"failed_op"}
@@ -116,7 +117,7 @@ def test_ray_executor_local_failing(local_ray_address: str, dagster_instance: Da
 RUNTIME_ENV = {"pip": ["polars"], "env_vars": {"FOO": "bar"}}
 
 
-def runtime_env_checks():
+def runtime_env_checks() -> None:
     import polars  # noqa  # type: ignore
     import os
 
@@ -124,16 +125,16 @@ def runtime_env_checks():
 
 
 @op
-def op_testing_runtime_env():
+def op_testing_runtime_env() -> None:
     runtime_env_checks()
 
 
 @job(executor_def=ray_executor)
-def job_testing_runtime_env():
+def job_testing_runtime_env() -> None:
     op_testing_runtime_env()
 
 
-def test_ray_executor_local_runtime_env(local_ray_address: str, dagster_instance: DagsterInstance):
+def test_ray_executor_local_runtime_env(local_ray_address: str, dagster_instance: DagsterInstance) -> None:
     # first test this runtime_env just with ray
 
     import ray
@@ -147,24 +148,26 @@ def test_ray_executor_local_runtime_env(local_ray_address: str, dagster_instance
             "execution": {
                 "config": {
                     "ray": {"address": local_ray_address, "runtime_env": RUNTIME_ENV},
-                }
-            }
+                },
+            },
         },
     )
     assert result.success, result.get_step_failure_events()[0].event_specific_data
 
 
 @op(tags={"dagster-ray/config": {"runtime_env": RUNTIME_ENV}})
-def op_with_user_provided_runtime_env():
+def op_with_user_provided_runtime_env() -> None:
     runtime_env_checks()
 
 
 @job(executor_def=ray_executor)
-def job_with_user_provided_runtime_env():
+def job_with_user_provided_runtime_env() -> None:
     op_with_user_provided_runtime_env()
 
 
-def test_ray_executor_local_user_provided_runtime_env(local_ray_address: str, dagster_instance: DagsterInstance):
+def test_ray_executor_local_user_provided_runtime_env(
+    local_ray_address: str, dagster_instance: DagsterInstance
+) -> None:
     # first test this runtime_env just with ray
 
     import ray
@@ -178,8 +181,8 @@ def test_ray_executor_local_user_provided_runtime_env(local_ray_address: str, da
             "execution": {
                 "config": {
                     "ray": {"address": local_ray_address},
-                }
-            }
+                },
+            },
         },
     )
     assert result.success, result.get_step_failure_events()[0].event_specific_data
diff --git a/tests/test_io_manager.py b/tests/test_io_manager.py
index 711e1c8e..7c5989cc 100644
--- a/tests/test_io_manager.py
+++ b/tests/test_io_manager.py
@@ -3,9 +3,9 @@
 from dagster_ray import RayIOManager
 
 
-def test_ray_io_manager():
+def test_ray_io_manager() -> None:
     @asset
-    def upstream():
+    def upstream() -> int:
         return 1
 
     @asset
@@ -18,7 +18,7 @@ def downstream(upstream) -> None:
     )
 
 
-def test_ray_io_manager_partitioned():
+def test_ray_io_manager_partitioned() -> None:
     partitions_def = StaticPartitionsDefinition(partition_keys=["A", "B", "C"])
 
     @asset(partitions_def=partitions_def)
@@ -37,7 +37,7 @@ def downstream_partitioned(context: AssetExecutionContext, upsteram_partitioned:
         )
 
 
-def test_ray_io_manager_partition_mapping():
+def test_ray_io_manager_partition_mapping() -> None:
     partitions_def = StaticPartitionsDefinition(partition_keys=["A", "B", "C"])
 
     @asset(partitions_def=partitions_def)
@@ -52,5 +52,6 @@ def downstream_non_partitioned(upsteram_partitioned: dict[str, str]) -> None:
         materialize([upsteram_partitioned], resources={"io_manager": RayIOManager()}, partition_key=partition_key)
 
     materialize(
-        [upsteram_partitioned.to_source_asset(), downstream_non_partitioned], resources={"io_manager": RayIOManager()}
+        [upsteram_partitioned.to_source_asset(), downstream_non_partitioned],
+        resources={"io_manager": RayIOManager()},
     )
diff --git a/tests/test_pipes.py b/tests/test_pipes.py
index a93186d5..18d02b0e 100644
--- a/tests/test_pipes.py
+++ b/tests/test_pipes.py
@@ -20,10 +20,10 @@ def pipes_ray_job_client(local_ray_address: str) -> PipesRayJobClient:
     return PipesRayJobClient(client=JobSubmissionClient(address=local_ray_address))
 
 
-def test_ray_job_pipes(pipes_ray_job_client: PipesRayJobClient, capsys):
+def test_ray_job_pipes(pipes_ray_job_client: PipesRayJobClient, capsys) -> None:
     @asset
     def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobClient):
-        result = pipes_ray_job_client.run(
+        return pipes_ray_job_client.run(
             context=context,
             submit_job_params={
                 "entrypoint": f"{sys.executable} {LOCAL_SCRIPT_PATH}",
@@ -31,8 +31,6 @@ def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobCl
             extras={"foo": "bar"},
         ).get_materialize_result()
 
-        return result
-
     with instance_for_test() as instance:
         result = materialize(
             [my_asset],
@@ -42,9 +40,6 @@ def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobCl
 
         captured = capsys.readouterr()
 
-        print(captured.out)
-        print(captured.err, file=sys.stderr)
-
         mat_evts = result.get_asset_materialization_events()
 
         mat = instance.get_latest_materialization_event(my_asset.key)
@@ -54,7 +49,8 @@ def my_asset(context: AssetExecutionContext, pipes_ray_job_client: PipesRayJobCl
 
         assert result.success
         assert mat
-        assert mat and mat.asset_materialization
+        assert mat
+        assert mat.asset_materialization
         assert mat.asset_materialization.metadata["some_metric"].value == 0
         assert mat.asset_materialization.tags
         assert mat.asset_materialization.tags[DATA_VERSION_TAG] == "alpha"
diff --git a/tests/utils.py b/tests/utils.py
index 207aac46..d4c11aff 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -2,9 +2,8 @@
 
 
 def get_saved_path(result: ExecuteInProcessResult, asset_name: str) -> str:
-    print("hello")
     path = (
-        list(filter(lambda evt: evt.is_handled_output, result.events_for_node(asset_name)))[0]
+        next(filter(lambda evt: evt.is_handled_output, result.events_for_node(asset_name)))
         .event_specific_data.metadata["path"]  # type: ignore
         .value
     )  # type: ignore[index,union-attr]