Apply ruff/pyupgrade rule UP031

Use format specifiers instead of percent format

Leave out tests for now.

Author: DimitriPapadopoulos

distributed/_concurrent_futures_thread.py

@@ -117,7 +117,7 @@ def __init__(self, max_workers=None, thread_name_prefix=""):
         self._shutdown = False
         self._shutdown_lock = threading.Lock()
         self._thread_name_prefix = thread_name_prefix or (
-            "ThreadPoolExecutor-%d" % next(self._counter)
+            f"ThreadPoolExecutor-{next(self._counter)}"
         )
 
     def submit(self, fn, *args, **kwargs):
@@ -144,9 +144,8 @@ def weakref_cb(_, q=self._work_queue):
         # idle threads than items in the work queue.
         num_threads = len(self._threads)
         if num_threads < self._max_workers:
-            thread_name = "%s_%d" % (self._thread_name_prefix or self, num_threads)
             t = threading.Thread(
-                name=thread_name,
+                name=f"{self._thread_name_prefix or self}_{num_threads}",
                 target=_worker,
                 args=(weakref.ref(self, weakref_cb), self._work_queue),
             )

distributed/batched.py

@@ -72,7 +72,7 @@ def __repr__(self):
         if self.closed():
             return "<BatchedSend: closed>"
         else:
-            return "<BatchedSend: %d in buffer>" % len(self.buffer)
+            return f"<BatchedSend: {len(self.buffer)} in buffer>"
 
     __str__ = __repr__
 

distributed/cfexecutor.py

@@ -56,8 +56,7 @@ def __init__(self, client, **kwargs):
         sk = set(kwargs)
         if not sk <= self._allowed_kwargs:
             raise TypeError(
-                "unsupported arguments to ClientExecutor: %s"
-                % sorted(sk - self._allowed_kwargs)
+                f"unsupported arguments to ClientExecutor: {sorted(sk - self._allowed_kwargs)}"
             )
         self._client = client
         self._futures = weakref.WeakSet()
@@ -127,7 +126,7 @@ def map(self, fn, *iterables, **kwargs):
             end_time = timeout + time()
         kwargs.pop("chunksize", None)
         if kwargs:
-            raise TypeError("unexpected arguments to map(): %s" % sorted(kwargs))
+            raise TypeError(f"unexpected arguments to map(): {sorted(kwargs)}")
 
         fs = self._client.map(fn, *iterables, **self._kwargs)
 

distributed/client.py

@@ -1378,17 +1378,8 @@ def __repr__(self):
         if addr:
             nworkers = info.get("n_workers", 0)
             nthreads = info.get("total_threads", 0)
-            text = "<%s: %r processes=%d threads=%d" % (
-                self.__class__.__name__,
-                addr,
-                nworkers,
-                nthreads,
-            )
             memory = info.get("total_memory", 0)
-            text += ", memory=" + format_bytes(memory)
-            text += ">"
-            return text
-
+            return f"<{self.__class__.__name__}: {addr!r} processes={nworkers} threads={nthreads}, memory={format_bytes(memory)}>"
         elif self.scheduler is not None:
             return f"<{self.__class__.__name__}: scheduler={self.scheduler.address!r}>"
         else:
@@ -2433,7 +2424,7 @@ async def wait(k):
                         bad_keys.add(key)
                         bad_data[key] = None
                     else:  # pragma: no cover
-                        raise ValueError("Bad value, `errors=%s`" % errors)
+                        raise ValueError(f"Bad value, `errors={errors}`")
 
             keys = [k for k in keys if k not in bad_keys and k not in data]
 
@@ -5886,7 +5877,7 @@ def update(self, futures):
         with self.lock:
             for f in futures:
                 if not isinstance(f, (Future, BaseActorFuture)):
-                    raise TypeError("Input must be a future, got %s" % f)
+                    raise TypeError(f"Input must be a future, got {f}")
                 self.futures[f] += 1
                 self.loop.add_callback(self._track_future, f)
 

distributed/comm/addressing.py

@@ -18,13 +18,13 @@ def parse_address(addr: str, strict: bool = False) -> tuple[str, str]:
     If strict is set to true the address must have a scheme.
     """
     if not isinstance(addr, str):
-        raise TypeError("expected str, got %r" % addr.__class__.__name__)
+        raise TypeError(f"expected str, got {addr.__class__.__name__!r}")
     scheme, sep, loc = addr.rpartition("://")
     if strict and not sep:
         msg = (
             "Invalid url scheme. "
             "Must include protocol like tcp://localhost:8000. "
-            "Got %s" % addr
+            f"Got {addr}"
         )
         raise ValueError(msg)
     if not sep:

distributed/comm/inproc.py

@@ -63,7 +63,7 @@ def get_listener_for(self, addr):
             return self.listeners.get(addr)
 
     def new_address(self):
-        return "%s/%d/%s" % (self.ip, os.getpid(), next(self.addr_suffixes))
+        return f"{self.ip}/{os.getpid()}/{next(self.addr_suffixes)}"
 
     def validate_address(self, addr):
         """
@@ -72,8 +72,7 @@ def validate_address(self, addr):
         ip, pid, suffix = addr.split("/")
         if ip != self.ip or int(pid) != os.getpid():
             raise ValueError(
-                "inproc address %r does not match host (%r) or pid (%r)"
-                % (addr, self.ip, os.getpid())
+                f"inproc address {addr!r} does not match host ({self.ip!r}) or pid ({os.getpid()!r})"
             )
 
 

distributed/comm/tcp.py

@@ -468,7 +468,7 @@ def _check_encryption(self, address, connection_args):
             # XXX Should we have a dedicated SecurityError class?
             raise RuntimeError(
                 "encryption required by Dask configuration, "
-                "refusing communication from/to %r" % (self.prefix + address,)
+                f"refusing communication from/to {self.prefix + address!r}"
             )
 
 

distributed/config.py

@@ -209,8 +209,9 @@ def get_loop_factory() -> Callable[[], asyncio.AbstractEventLoop] | None:
             return asyncio.SelectorEventLoop
         return None
     raise ValueError(
-        "Expected distributed.admin.event-loop to be in ('asyncio', 'tornado', 'uvloop'), got %s"
-        % dask.config.get("distributed.admin.event-loop")
+        "Expected distributed.admin.event-loop to be in ('asyncio', 'tornado', 'uvloop'), got {}".format(
+            dask.config.get("distributed.admin.event-loop")
+        )
     )
 
 

distributed/core.py

@@ -291,7 +291,7 @@ def __init__(
         self.handlers.update(handlers)
         if blocked_handlers is None:
             blocked_handlers = dask.config.get(
-                "distributed.%s.blocked-handlers" % type(self).__name__.lower(), []
+                f"distributed.{type(self).__name__.lower()}.blocked-handlers", []
             )
         self.blocked_handlers = blocked_handlers
         self.stream_handlers = {}
@@ -1227,7 +1227,7 @@ def __del__(self):
                     comm.abort()
 
     def __repr__(self):
-        return "<rpc to %r, %d comms>" % (self.address, len(self.comms))
+        return f"<rpc to {self.address!r}, {len(self.comms)} comms>"
 
 
 class PooledRPCCall:
@@ -1378,11 +1378,7 @@ def open(self) -> int:
         return self.active + sum(map(len, self.available.values()))
 
     def __repr__(self) -> str:
-        return "<ConnectionPool: open=%d, active=%d, connecting=%d>" % (
-            self.open,
-            self.active,
-            len(self._connecting),
-        )
+        return f"<ConnectionPool: open={self.open}, active={self.active}, connecting={len(self._connecting)}>"
 
     def __call__(
         self,

distributed/dashboard/components/scheduler.py

@@ -3104,15 +3104,15 @@ def update(self):
                 #      (ones without any compute during the relevant dt)
                 #   3. Colors the labels appropriately.
                 formatter = CustomJSHover(
-                    code="""
-                        const colormap = %s;
+                    code=f"""
+                        const colormap = {dict(zip(stackers, colors))};
                         const divs = [];
-                        for (let k of Object.keys(source.data)) {
+                        for (let k of Object.keys(source.data)) {{
                           const val = source.data[k][value];
                           const color = colormap[k];
-                          if (k === "time" || k === "nthreads" || val < 1.e-3) {
+                          if (k === "time" || k === "nthreads" || val < 1.e-3) {{
                             continue;
-                          }
+                          }}
                           const label = k.length >= 20 ? k.slice(0, 20) + '…' : k;
 
                           // Unshift so that the ordering of the labels is the same as
@@ -3127,18 +3127,15 @@ def update(self):
                               + '</div>'
                           )
 
-                        }
+                        }}
                         divs.unshift(
                           '<div>'
                             + '<span style="font-weight: bold; color: darkgrey;">nthreads: </span>'
                             + source.data.nthreads[value]
                             + '</div>'
                         );
                         return divs.join('\\n')
-                        """
-                    % dict(
-                        zip(stackers, colors)
-                    ),  # sneak the color mapping into the callback
+                        """,  # sneak the color mapping into the callback
                     args={"source": self.source},
                 )
                 # Add the HoverTool to the top line renderer.
@@ -3378,13 +3375,13 @@ def update(self):
         )
 
         self.root.title.text = (
-            "Progress -- total: %(all)s, "
-            "waiting: %(waiting)s, "
-            "queued: %(queued)s, "
-            "processing: %(processing)s, "
-            "in-memory: %(memory)s, "
-            "no-worker: %(no_worker)s, "
-            "erred: %(erred)s" % totals
+            "Progress -- total: {all}, "
+            "waiting: {waiting}, "
+            "queued: {queued}, "
+            "processing: {processing}, "
+            "in-memory: {memory}, "
+            "no-worker: {no_worker}, "
+            "erred: {erred}".format(**totals)
         )
 
 
@@ -4537,8 +4534,9 @@ def __init__(self, scheduler, start=None):
         else:
             logs_html = Log(
                 "\n".join(
-                    "%s - %s"
-                    % (datetime.fromtimestamp(time).strftime("%H:%M:%S.%f"), line)
+                    "{} - {}".format(
+                        datetime.fromtimestamp(time).strftime("%H:%M:%S.%f"), line
+                    )
                     for time, level, line in logs
                 )
             )._repr_html_()