Enforce ruff rules (RUF) (#9153)

Author: DimitriPapadopoulos

distributed/cfexecutor.py

@@ -125,8 +125,7 @@ def map(self, fn, *iterables, **kwargs):
         if timeout is not None:
             timeout = parse_timedelta(timeout)
             end_time = timeout + time()
-        if "chunksize" in kwargs:
-            del kwargs["chunksize"]
+        kwargs.pop("chunksize", None)
         if kwargs:
             raise TypeError("unexpected arguments to map(): %s" % sorted(kwargs))
 

distributed/client.py

@@ -2297,8 +2297,8 @@ def map(
                 keys = [list(element) for element in partition_all(batch_size, key)]
             else:
                 keys = [key for _ in range(len(batches))]
-            return sum(
-                (
+            return list(
+                flatten(
                     self.map(
                         func,
                         *batch,
@@ -2315,8 +2315,7 @@ def map(
                         **kwargs,
                     )
                     for key, batch in zip(keys, batches)
-                ),
-                [],
+                )
             )
 
         key = key or funcname(func)

distributed/core.py

@@ -520,7 +520,7 @@ async def start(self):
             timeout = getattr(self, "death_timeout", None)
 
             async def _close_on_failure(exc: Exception) -> None:
-                await self.close(reason=f"failure-to-start-{str(type(exc))}")
+                await self.close(reason=f"failure-to-start-{type(exc)}")
                 self.status = Status.failed
                 self.__startup_exc = exc
 

distributed/deploy/local.py

@@ -211,7 +211,7 @@ def __init__(
             n_workers = max(1, CPU_COUNT // threads_per_worker) if processes else 1
         if n_workers and threads_per_worker is None:
             # Overcommit threads per worker, rather than undercommit
-            threads_per_worker = max(1, int(math.ceil(CPU_COUNT / n_workers)))
+            threads_per_worker = max(1, math.ceil(CPU_COUNT / n_workers))
         if n_workers and "memory_limit" not in worker_kwargs:
             worker_kwargs["memory_limit"] = parse_memory_limit(
                 "auto", 1, n_workers, logger=logger

distributed/deploy/spec.py

@@ -514,10 +514,10 @@ def _memory_per_worker(self) -> int:
 
     def scale(self, n=0, memory=None, cores=None):
         if memory is not None:
-            n = max(n, int(math.ceil(parse_bytes(memory) / self._memory_per_worker())))
+            n = max(n, math.ceil(parse_bytes(memory) / self._memory_per_worker()))
 
         if cores is not None:
-            n = max(n, int(math.ceil(cores / self._threads_per_worker())))
+            n = max(n, math.ceil(cores / self._threads_per_worker()))
 
         if len(self.worker_spec) > n:
             not_yet_launched = set(self.worker_spec) - {

distributed/deploy/subprocess.py

@@ -241,7 +241,7 @@ def SubprocessCluster(
         n_workers = max(1, CPU_COUNT // threads_per_worker)
     if n_workers and threads_per_worker is None:
         # Overcommit threads per worker, rather than undercommit
-        threads_per_worker = max(1, int(math.ceil(CPU_COUNT / n_workers)))
+        threads_per_worker = max(1, math.ceil(CPU_COUNT / n_workers))
     if n_workers and "memory_limit" not in worker_kwargs:
         worker_kwargs["memory_limit"] = parse_memory_limit(
             "auto", 1, n_workers, logger=logger

distributed/protocol/serialize.py

@@ -126,7 +126,7 @@ def import_allowed_module(name):
         return _cached_allowed_modules[name]
     else:
         raise RuntimeError(
-            f"Importing {repr(name)} is not allowed, please add it to the list of "
+            f"Importing {name!r} is not allowed, please add it to the list of "
             "allowed modules the scheduler can import via the "
             "distributed.scheduler.allowed-imports configuration setting."
         )

distributed/protocol/tests/test_serialize.py

@@ -284,8 +284,8 @@ def test_serialize_bytes(kwargs):
         1,
         "abc",
         b"ab" * int(40e6),
-        int(2**26) * b"ab",
-        (int(2**25) * b"ab", int(2**25) * b"ab"),
+        (2**26) * b"ab",
+        ((2**25) * b"ab", (2**25) * b"ab"),
     ]:
         b = serialize_bytes(x, **kwargs)
         assert isinstance(b, bytes)

distributed/shuffle/tests/test_buffer.py

@@ -15,7 +15,7 @@
 
 
 def gen_bytes(percentage: float, limit: int) -> bytes:
-    num_bytes = int(math.floor(percentage * limit))
+    num_bytes = math.floor(percentage * limit)
     return b"0" * num_bytes
 
 

distributed/shuffle/tests/test_comm_buffer.py

@@ -94,7 +94,7 @@ async def send(address, shards):
 
 
 def gen_bytes(percentage: float, memory_limit: int) -> bytes:
-    num_bytes = int(math.floor(percentage * memory_limit))
+    num_bytes = math.floor(percentage * memory_limit)
     return b"0" * num_bytes