Style: update flake8 to latest

.flake8

@@ -44,6 +44,8 @@ ignore =
   B015
   B016
   B017
+  B023
+  B026
 avoid-escape = no
 # Error E731 is ignored because of the migration from YAPF to Black.
 # See https://github.com/ray-project/ray/issues/21315 for more information.

.pre-commit-config.yaml

@@ -47,14 +47,14 @@ repos:
         types_or: [python]
 
   - repo: https://github.com/pycqa/flake8
-    rev: 3.9.1
+    rev: 7.1.1
     hooks:
       - id: flake8
         additional_dependencies:
           [
-            flake8-comprehensions==3.10.1,
-            flake8-quotes==2.0.0,
-            flake8-bugbear==21.9.2,
+            flake8-comprehensions==3.15.0,
+            flake8-quotes==3.4.0,
+            flake8-bugbear==24.8.19,
           ]
 
   - repo: https://github.com/pre-commit/mirrors-prettier

ci/lint/format.sh

@@ -5,7 +5,7 @@
 # Cause the script to exit if a single command fails
 set -euo pipefail
 
-FLAKE8_VERSION_REQUIRED="3.9.1"
+FLAKE8_VERSION_REQUIRED="7.1.1"
 BLACK_VERSION_REQUIRED="22.10.0"
 SHELLCHECK_VERSION_REQUIRED="0.7.1"
 MYPY_VERSION_REQUIRED="1.7.0"

doc/source/custom_directives.py

@@ -776,8 +776,8 @@ def render_library_examples(config: pathlib.Path = None) -> bs4.BeautifulSoup:
         soup.append(page_text)
 
         container = soup.new_tag("div", attrs={"class": "example-index"})
-        for group, examples in examples.items():
-            if not examples:
+        for group, _examples in examples.items():
+            if not _examples:
                 continue
 
             header = soup.new_tag("h2", attrs={"class": "example-header"})
@@ -810,7 +810,7 @@ def render_library_examples(config: pathlib.Path = None) -> bs4.BeautifulSoup:
                 table.append(thead)
 
             tbody = soup.new_tag("tbody")
-            for example in examples:
+            for _example in _examples:
                 tr = soup.new_tag("tr")
 
                 # The columns specify which attributes of each example to show;
@@ -821,7 +821,7 @@ def render_library_examples(config: pathlib.Path = None) -> bs4.BeautifulSoup:
                         col_td = soup.new_tag("td")
                         col_p = soup.new_tag("p")
 
-                        attribute_value = getattr(example, attribute, "")
+                        attribute_value = getattr(_example, attribute, "")
                         if isinstance(attribute_value, str):
                             col_p.append(attribute_value)
                         elif isinstance(attribute_value, list):
@@ -834,14 +834,14 @@ def render_library_examples(config: pathlib.Path = None) -> bs4.BeautifulSoup:
 
                 link_td = soup.new_tag("td")
                 link_p = soup.new_tag("p")
-                if example.link.startswith("http"):
-                    link_href = soup.new_tag("a", attrs={"href": example.link})
+                if _example.link.startswith("http"):
+                    link_href = soup.new_tag("a", attrs={"href": _example.link})
                 else:
                     link_href = soup.new_tag(
-                        "a", attrs={"href": context["pathto"](example.link)}
+                        "a", attrs={"href": context["pathto"](_example.link)}
                     )
                 link_span = soup.new_tag("span")
-                link_span.append(example.title)
+                link_span.append(_example.title)
                 link_href.append(link_span)
                 link_p.append(link_href)
                 link_td.append(link_p)

doc/source/serve/doc_code/tutorial_sklearn.py

@@ -28,7 +28,8 @@
     iris_dataset["target_names"],
 )
 
-np.random.shuffle(data), np.random.shuffle(target)
+np.random.shuffle(data)
+np.random.shuffle(target)
 train_x, train_y = data[:100], target[:100]
 val_x, val_y = data[100:], target[100:]
 # __doc_data_end__

python/ray/_private/accelerators/accelerator.py

@@ -86,7 +86,9 @@ def validate_resource_request_quantity(
     @staticmethod
     @abstractmethod
     def get_current_process_visible_accelerator_ids() -> Optional[List[str]]:
-        """Get the ids of accelerators of this family that are visible to the current process.
+        """
+        Get the ids of accelerators of this family
+        that are visible to the current process.
 
         Returns:
             The list of visiable accelerator ids.
@@ -96,7 +98,9 @@ def get_current_process_visible_accelerator_ids() -> Optional[List[str]]:
     @staticmethod
     @abstractmethod
     def set_current_process_visible_accelerator_ids(ids: List[str]) -> None:
-        """Set the ids of accelerators of this family that are visible to the current process.
+        """
+        Set the ids of accelerators of this family
+        that are visible to the current process.
 
         Args:
             ids: The ids of visible accelerators of this family.
@@ -106,7 +110,8 @@ def set_current_process_visible_accelerator_ids(ids: List[str]) -> None:
     def get_ec2_instance_num_accelerators(
         instance_type: str, instances: dict
     ) -> Optional[int]:
-        """Get the number of accelerators of this family on ec2 instance with given type.
+        """
+        Get the number of accelerators of this family on ec2 instance with given type.
 
         Args:
             instance_type: The ec2 instance type.

python/ray/_private/node.py

@@ -534,7 +534,8 @@ def get_resource_spec(self):
         """Resolve and return the current resource spec for the node."""
 
         def merge_resources(env_dict, params_dict):
-            """Separates special case params and merges two dictionaries, picking from the
+            """
+            Separates special case params and merges two dictionaries, picking from the
             first in the event of a conflict. Also emit a warning on every
             conflict.
             """

python/ray/_private/runtime_env/plugin.py

@@ -1,7 +1,7 @@
 import logging
 import os
 import json
-from abc import ABC
+from abc import ABC, abstractmethod
 from typing import List, Dict, Optional, Any, Type
 
 from ray._private.runtime_env.context import RuntimeEnvContext
@@ -28,6 +28,7 @@ class RuntimeEnvPlugin(ABC):
     priority: int = RAY_RUNTIME_ENV_PLUGIN_DEFAULT_PRIORITY
 
     @staticmethod
+    @abstractmethod
     def validate(runtime_env_dict: dict) -> None:
         """Validate user entry for this plugin.
 

python/ray/_private/runtime_env/utils.py

@@ -93,7 +93,7 @@ async def check_output_cmd(
         # since Python 3.9, when cancelled, the inner process needs to throw as it is
         # for asyncio to timeout properly https://bugs.python.org/issue40607
         raise e
-    except BaseException as e:
+    except BaseException as e:  # noqa: B036 To avoid breaking change
         raise RuntimeError(f"Run cmd[{cmd_index}] got exception.") from e
     else:
         stdout = stdout.decode("utf-8")

python/ray/_private/services.py

@@ -1608,7 +1608,7 @@ def start_raylet(
     Returns:
         ProcessInfo for the process that was started.
     """
-    assert node_manager_port is not None and type(node_manager_port) == int
+    assert node_manager_port is not None and isinstance(node_manager_port, int)
 
     if use_valgrind and use_profiler:
         raise ValueError("Cannot use valgrind and profiler at the same time.")

python/ray/_private/test_utils.py

@@ -790,7 +790,7 @@ def wait_until_succeeded_without_exception(
     Return:
         Whether exception occurs within a timeout.
     """
-    if type(exceptions) != tuple:
+    if not isinstance(exceptions, tuple):
         raise Exception("exceptions arguments should be given as a tuple")
 
     time_elapsed = 0

python/ray/_private/worker.py

@@ -1220,13 +1220,18 @@ def __getitem__(self, key):
 
     def __len__(self):
         if log_once("ray_context_len"):
-            warnings.warn("len(ctx) is deprecated. Use len(ctx.address_info) instead.")
+            warnings.warn(
+                "len(ctx) is deprecated. Use len(ctx.address_info) instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
         return len(self.address_info)
 
     def __iter__(self):
         if log_once("ray_context_len"):
             warnings.warn(
-                "iter(ctx) is deprecated. Use iter(ctx.address_info) instead."
+                "iter(ctx) is deprecated. Use iter(ctx.address_info) instead.",
+                stacklevel=2,
             )
         return iter(self.address_info)
 
@@ -2762,9 +2767,9 @@ def get(
                 port=None,
                 patch_stdstreams=False,
                 quiet=None,
-                breakpoint_uuid=debugger_breakpoint.decode()
-                if debugger_breakpoint
-                else None,
+                breakpoint_uuid=(
+                    debugger_breakpoint.decode() if debugger_breakpoint else None
+                ),
                 debugger_external=worker.ray_debugger_external,
             )
             rdb.set_trace(frame=frame)

python/ray/air/_internal/device_manager/torch_device_manager.py

@@ -1,4 +1,4 @@
-from abc import ABC
+from abc import ABC, abstractmethod
 from typing import List, Union
 
 import torch
@@ -9,32 +9,39 @@ class TorchDeviceManager(ABC):
     an acclerator family in Ray AI Library.
     """
 
+    @abstractmethod
     def is_available(self) -> bool:
         """Validate if device is available."""
         ...
 
+    @abstractmethod
     def get_devices(self) -> List[torch.device]:
         """Gets the correct torch device configured for this process"""
         ...
 
+    @abstractmethod
     def set_device(self, device: Union[torch.device, int, str, None]):
         """Set the correct device for this process"""
         ...
 
+    @abstractmethod
     def supports_stream(self) -> bool:
         """Validate if the device type support create a stream"""
         ...
 
+    @abstractmethod
     def create_stream(self, device: torch.device):
         """Create a device stream"""
         ...
 
+    @abstractmethod
     def get_stream_context(self, stream):
         """Get a stream context of device. If device didn't support stream,
         this should return a empty context manager instead of  None.
         """
         ...
 
+    @abstractmethod
     def get_current_stream(self):
         """Get current stream on accelerators like torch.cuda.current_stream"""
         ...

python/ray/air/_internal/util.py

@@ -122,7 +122,7 @@ def run(self):
             else:
                 # If non-zero exit code, then raise exception to main thread.
                 self._propagate_exception(e)
-        except BaseException as e:
+        except BaseException as e:  # noqa: B036 To avoid breaking change
             # Propagate all other exceptions to the main thread.
             self._propagate_exception(e)
 

python/ray/air/execution/resources/resource_manager.py

@@ -62,6 +62,7 @@ class ResourceManager(abc.ABC):
 
     """
 
+    @abc.abstractmethod
     def request_resources(self, resource_request: ResourceRequest):
         """Request resources.
 
@@ -75,6 +76,7 @@ def request_resources(self, resource_request: ResourceRequest):
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
     def cancel_resource_request(self, resource_request: ResourceRequest):
         """Cancel resource request.
 
@@ -84,10 +86,12 @@ def cancel_resource_request(self, resource_request: ResourceRequest):
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
     def has_resources_ready(self, resource_request: ResourceRequest) -> bool:
         """Returns True if resources for the given request are ready to be acquired."""
         raise NotImplementedError
 
+    @abc.abstractmethod
     def acquire_resources(
         self, resource_request: ResourceRequest
     ) -> Optional[AcquiredResources]:
@@ -98,6 +102,7 @@ def acquire_resources(
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
     def free_resources(self, acquired_resource: AcquiredResources):
         """Free acquired resources from usage and return them to the resource manager.
 
@@ -108,6 +113,7 @@ def free_resources(self, acquired_resource: AcquiredResources):
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
     def get_resource_futures(self) -> List[ray.ObjectRef]:
         """Return futures for resources to await.
 
@@ -120,6 +126,7 @@ def get_resource_futures(self) -> List[ray.ObjectRef]:
         """
         return []
 
+    @abc.abstractmethod
     def update_state(self):
         """Update internal state of the resource manager.
 
@@ -132,6 +139,7 @@ def update_state(self):
         """
         pass
 
+    @abc.abstractmethod
     def clear(self):
         """Reset internal state and clear all resources.
 

python/ray/air/integrations/wandb.py

@@ -558,6 +558,7 @@ def __init__(
             warnings.warn(
                 "`save_checkpoints` is deprecated. Use `upload_checkpoints` instead.",
                 DeprecationWarning,
+                stacklevel=2,
             )
             upload_checkpoints = save_checkpoints
 

python/ray/air/tests/test_integration_comet.py

@@ -186,7 +186,7 @@ def test_kwargs_passthrough(self, experiment):
         logger.log_trial_start(trial)
 
         # These are the default kwargs that get passed to create the experiment
-        expected_kwargs = {kwarg: False for kwarg in logger._exclude_autolog}
+        expected_kwargs = dict.fromkeys(logger._exclude_autolog, False)
         expected_kwargs.update(experiment_kwargs)
 
         experiment.assert_called_with(**expected_kwargs)

python/ray/air/tests/test_integration_wandb.py

@@ -288,7 +288,7 @@ def test_wandb_logger_reporting(self, trial):
     def test_wandb_logger_auto_config_keys(self, trial):
         logger = WandbTestExperimentLogger(project="test_project", api_key="1234")
         logger.on_trial_start(iteration=0, trials=[], trial=trial)
-        result = {key: 0 for key in WandbLoggerCallback.AUTO_CONFIG_KEYS}
+        result = dict.fromkeys(WandbLoggerCallback.AUTO_CONFIG_KEYS, 0)
         logger.on_trial_result(0, [], trial, result)
         logger.on_trial_complete(0, [], trial)
         logger.on_experiment_end(trials=[trial])
@@ -314,7 +314,7 @@ def test_wandb_logger_exclude_config(self):
         logger.on_trial_start(iteration=0, trials=[], trial=trial)
 
         # We need to test that `excludes` also applies to `AUTO_CONFIG_KEYS`.
-        result = {key: 0 for key in WandbLoggerCallback.AUTO_CONFIG_KEYS}
+        result = dict.fromkeys(WandbLoggerCallback.AUTO_CONFIG_KEYS, 0)
         logger.on_trial_result(0, [], trial, result)
         logger.on_trial_complete(0, [], trial)
         logger.on_experiment_end(trials=[trial])

python/ray/air/util/data_batch_conversion.py

@@ -158,6 +158,7 @@ def convert_batch_type_to_pandas(
         "starting from Ray 2.4. All batch format conversions should be "
         "done manually instead of relying on this API.",
         PendingDeprecationWarning,
+        stacklevel=2,
     )
     return _convert_batch_type_to_pandas(
         data=data, cast_tensor_columns=cast_tensor_columns
@@ -186,6 +187,7 @@ def convert_pandas_to_batch_type(
         "starting from Ray 2.4. All batch format conversions should be "
         "done manually instead of relying on this API.",
         PendingDeprecationWarning,
+        stacklevel=2,
     )
     return _convert_pandas_to_batch_type(
         data=data, type=type, cast_tensor_columns=cast_tensor_columns

python/ray/air/util/torch_dist.py

@@ -6,7 +6,7 @@
 """
 
 import os
-from abc import ABC
+from abc import ABC, abstractmethod
 from collections import defaultdict
 from datetime import timedelta
 from typing import Callable, List, T
@@ -27,6 +27,7 @@ class TorchDistributedWorker(ABC):
     to be executed on a remote DDP worker.
     """
 
+    @abstractmethod
     def execute(self, func: Callable[..., T], *args, **kwargs) -> T:
         """Executes the input function and returns the output.