LINT: Fixup black string normalization

distributed/cli/dask_scheduler.py

@@ -215,7 +215,7 @@ def del_pid_file():
         port=port,
         dashboard_address=dashboard_address if dashboard else None,
         service_kwargs={"dashboard": {"prefix": dashboard_prefix}},
-        **kwargs,
+        **kwargs
     )
     logger.info("Local Directory: %26s", local_directory)
     logger.info("-" * 47)

distributed/cli/dask_worker.py

@@ -84,7 +84,7 @@
     "--listen-address",
     type=str,
     default=None,
-    help="The address to which the worker binds. " "Example: tcp://0.0.0.0:9000",
+    help="The address to which the worker binds. Example: tcp://0.0.0.0:9000",
 )
 @click.option(
     "--contact-address",

distributed/client.py

@@ -905,7 +905,7 @@ async def _start(self, timeout=no_default, **kwargs):
                 pass
             except Exception:
                 logger.info(
-                    "Tried to start cluster and received an error. " "Proceeding.",
+                    "Tried to start cluster and received an error. Proceeding.",
                     exc_info=True,
                 )
             address = self.cluster.scheduler_address
@@ -2383,9 +2383,7 @@ def _graph_to_futures(
             dsk3 = {k: v for k, v in dsk2.items() if k is not v}
             for future in extra_futures:
                 if future.client is not self:
-                    msg = (
-                        "Inputs contain futures that were created by " "another client."
-                    )
+                    msg = "Inputs contain futures that were created by another client."
                     raise ValueError(msg)
 
             if restrictions:
@@ -3485,7 +3483,7 @@ def to_packages(d):
                     errs.append("%s\n%s" % (pkg, asciitable(["", "version"], rows)))
 
                 raise ValueError(
-                    "Mismatched versions found\n" "\n" "%s" % ("\n\n".join(errs))
+                    "Mismatched versions found\n\n%s" % ("\n\n".join(errs))
                 )
 
         return result
@@ -3967,7 +3965,7 @@ async def _wait(fs, timeout=None, return_when=ALL_COMPLETED):
         wait_for = Any
     else:
         raise NotImplementedError(
-            "Only return_when='ALL_COMPLETED' and 'FIRST_COMPLETED' are " "supported"
+            "Only return_when='ALL_COMPLETED' and 'FIRST_COMPLETED' are supported"
         )
 
     future = wait_for({f._state.wait() for f in fs})

distributed/comm/addressing.py

@@ -123,7 +123,7 @@ def get_address_host_port(addr, strict=False):
         return backend.get_address_host_port(loc)
     except NotImplementedError:
         raise ValueError(
-            "don't know how to extract host and port " "for address %r" % (addr,)
+            "don't know how to extract host and port for address %r" % (addr,)
         )
 
 

distributed/comm/tests/test_ucx.py

@@ -261,7 +261,7 @@ def test_ucx_localcluster(loop, processes):
         threads_per_worker=1,
         processes=processes,
         loop=loop,
-        **kwargs,
+        **kwargs
     ) as cluster:
         with Client(cluster) as client:
             x = client.submit(inc, 1)

distributed/diskutils.py

@@ -237,7 +237,7 @@ def new_work_dir(self, **kwargs):
             self._purge_leftovers()
         except OSError:
             logger.error(
-                "Failed to clean up lingering worker directories " "in path: %s ",
+                "Failed to clean up lingering worker directories in path: %s ",
                 exc_info=True,
             )
         return WorkDir(self, **kwargs)

distributed/protocol/keras.py

@@ -9,7 +9,7 @@ def serialize_keras_model(model):
 
     if keras.__version__ < "1.2.0":
         raise ImportError(
-            "Need Keras >= 1.2.0. " "Try pip install keras --upgrade --no-deps"
+            "Need Keras >= 1.2.0. Try pip install keras --upgrade --no-deps"
         )
 
     header = model._updated_config()

distributed/scheduler.py

@@ -2366,9 +2366,7 @@ def handle_long_running(self, key=None, worker=None, compute_duration=None):
 
         ws = ts.processing_on
         if ws is None:
-            logger.debug(
-                "Received long-running signal from duplicate task. " "Ignoring."
-            )
+            logger.debug("Received long-running signal from duplicate task. Ignoring.")
             return
 
         if compute_duration:
@@ -4730,7 +4728,7 @@ def check_worker_ttl(self):
         for ws in self.workers.values():
             if ws.last_seen < now - self.worker_ttl:
                 logger.warning(
-                    "Worker failed to heartbeat within %s seconds. " "Closing: %s",
+                    "Worker failed to heartbeat within %s seconds. Closing: %s",
                     self.worker_ttl,
                     ws,
                 )

distributed/tests/test_worker.py

@@ -464,7 +464,7 @@ def test_Executor(c, s):
 
 
 @pytest.mark.skip(
-    reason="Other tests leak memory, so process-level checks" "trigger immediately"
+    reason="Other tests leak memory, so process-level checks trigger immediately"
 )
 @gen_cluster(
     client=True,

distributed/utils_perf.py

@@ -42,7 +42,7 @@ def collect(self):
         elapsed = max(collect_start - self.last_collect, MIN_RUNTIME)
         if self.last_gc_duration / elapsed < self.max_in_gc_frac:
             self.logger.debug(
-                "Calling gc.collect(). %0.3fs elapsed since " "previous call.", elapsed
+                "Calling gc.collect(). %0.3fs elapsed since previous call.", elapsed
             )
             gc.collect()
             self.last_collect = collect_start

distributed/worker.py

@@ -3354,7 +3354,7 @@ async def run(server, comm, function, args=(), kwargs={}, is_coro=None, wait=Tru
 
     except Exception as e:
         logger.warning(
-            " Run Failed\n" "Function: %s\n" "args:     %s\n" "kwargs:   %s\n",
+            "Run Failed\nFunction: %s\nargs:     %s\nkwargs:   %s\n",
             str(funcname(function))[:1000],
             convert_args_to_str(args, max_len=1000),
             convert_kwargs_to_str(kwargs, max_len=1000),