Set float_to_top isort flag

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   -   repo: https://github.com/pycqa/isort
-      rev: 5.8.0
+      rev: 5.9.3
       hooks:
       - id: isort
         language_version: python3

conftest.py

@@ -1,6 +1,9 @@
 # https://pytest.org/latest/example/simple.html#control-skipping-of-tests-according-to-command-line-option
 import pytest
 
+# Make all fixtures available
+from distributed.utils_test import *  # noqa
+
 # Uncomment to enable more logging and checks
 # (https://docs.python.org/3/library/asyncio-dev.html)
 # Note this makes things slower and might consume much memory.
@@ -16,9 +19,6 @@
     except Exception:
         pass
 
-# Make all fixtures available
-from distributed.utils_test import *  # noqa
-
 
 def pytest_addoption(parser):
     parser.addoption("--runslow", action="store_true", help="run slow tests")

distributed/_ipython_utils.py

@@ -6,13 +6,7 @@
 
 import atexit
 import os
-
-try:
-    import queue
-except ImportError:
-    # Python 2
-    import Queue as queue
-
+import queue
 import sys
 from subprocess import Popen
 from threading import Event, Thread

distributed/cli/tests/test_dask_scheduler.py

@@ -1,15 +1,11 @@
-import psutil
-import pytest
-
-pytest.importorskip("requests")
-
 import os
 import shutil
 import sys
 import tempfile
 from time import sleep
 
-import requests
+import psutil
+import pytest
 from click.testing import CliRunner
 
 import distributed
@@ -24,6 +20,8 @@
     popen,
 )
 
+requests = pytest.importorskip("requests")
+
 
 def test_defaults(loop):
     with popen(["dask-scheduler"]):

distributed/cli/tests/test_dask_worker.py

@@ -1,15 +1,10 @@
 import asyncio
-
-import pytest
-from click.testing import CliRunner
-
-pytest.importorskip("requests")
-
 import os
 from multiprocessing import cpu_count
 from time import sleep
 
-import requests
+import pytest
+from click.testing import CliRunner
 
 import distributed.cli.dask_worker
 from distributed import Client, Scheduler
@@ -19,6 +14,8 @@
 from distributed.utils import parse_ports, sync, tmpfile
 from distributed.utils_test import gen_cluster, popen, terminate_process, wait_for_port
 
+requests = pytest.importorskip("requests")
+
 
 def test_nanny_worker_ports(loop):
     with popen(["dask-scheduler", "--port", "9359", "--no-dashboard"]):

distributed/client.py

@@ -25,10 +25,13 @@
 from queue import Queue as pyQueue
 
 from tlz import first, groupby, keymap, merge, partition_all, valmap
+from tornado import gen
+from tornado.ioloop import IOLoop, PeriodicCallback
 
 import dask
 from dask.base import collections_to_dsk, normalize_token, tokenize
 from dask.core import flatten
+from dask.delayed import single_key
 from dask.highlevelgraph import HighLevelGraph
 from dask.optimization import SubgraphCallable
 from dask.utils import (
@@ -41,13 +44,6 @@
     stringify,
 )
 
-try:
-    from dask.delayed import single_key
-except ImportError:
-    single_key = first
-from tornado import gen
-from tornado.ioloop import IOLoop, PeriodicCallback
-
 from . import versions as version_module
 from .batched import BatchedSend
 from .cfexecutor import ClientExecutor

distributed/comm/tcp.py

@@ -8,15 +8,8 @@
 import weakref
 from ssl import SSLCertVerificationError, SSLError
 
-from tornado import gen
-
-try:
-    import ssl
-except ImportError:
-    ssl = None
-
 from tlz import sliding_window
-from tornado import netutil
+from tornado import gen, netutil
 from tornado.iostream import StreamClosedError
 from tornado.tcpclient import TCPClient
 from tornado.tcpserver import TCPServer
@@ -33,6 +26,12 @@
 from .registry import Backend, backends
 from .utils import ensure_concrete_host, from_frames, get_tcp_server_address, to_frames
 
+try:
+    import ssl
+except ImportError:
+    ssl = None
+
+
 logger = logging.getLogger(__name__)
 
 

distributed/comm/tests/test_ucx.py

@@ -2,17 +2,18 @@
 
 import pytest
 
-pytestmark = pytest.mark.gpu
-
-ucp = pytest.importorskip("ucp")
-
 from distributed import Client, Scheduler, wait
 from distributed.comm import connect, listen, parse_address, ucx
 from distributed.comm.registry import backends, get_backend
 from distributed.deploy.local import LocalCluster
 from distributed.protocol import to_serialize
 from distributed.utils_test import gen_cluster, gen_test, inc
 
+pytestmark = pytest.mark.gpu
+
+ucp = pytest.importorskip("ucp")
+
+
 try:
     HOST = ucp.get_address()
 except Exception:

distributed/comm/tests/test_ucx_config.py

@@ -2,15 +2,16 @@
 
 import pytest
 
-pytestmark = pytest.mark.gpu
-
 import dask
 
 from distributed import Client
 from distributed.comm.ucx import _scrub_ucx_config
 from distributed.utils import get_ip
 from distributed.utils_test import popen
 
+pytestmark = pytest.mark.gpu
+
+
 try:
     HOST = get_ip()
 except Exception:

distributed/dashboard/components/scheduler.py

@@ -40,6 +40,7 @@
 from bokeh.plotting import figure
 from bokeh.themes import Theme
 from bokeh.transform import cumsum, factor_cmap, linear_cmap
+from jinja2 import Environment, FileSystemLoader
 from tlz import curry, pipe
 from tlz.curried import concat, groupby, map
 from tornado import escape
@@ -48,11 +49,6 @@
 from dask import config
 from dask.utils import format_bytes, format_time, key_split, parse_timedelta
 
-try:
-    import numpy as np
-except ImportError:
-    np = False
-
 from distributed.dashboard.components import add_periodic_callback
 from distributed.dashboard.components.shared import (
     DashboardComponent,
@@ -69,14 +65,19 @@
 from distributed.metrics import time
 from distributed.utils import Log, log_errors
 
+try:
+    import numpy as np
+except ImportError:
+    np = False
+
+
 if dask.config.get("distributed.dashboard.export-tool"):
     from distributed.dashboard.export_tool import ExportTool
 else:
     ExportTool = None
 
 logger = logging.getLogger(__name__)
 
-from jinja2 import Environment, FileSystemLoader
 
 env = Environment(
     loader=FileSystemLoader(

distributed/dashboard/components/worker.py

@@ -19,6 +19,7 @@
 from bokeh.palettes import RdBu
 from bokeh.plotting import figure
 from bokeh.themes import Theme
+from jinja2 import Environment, FileSystemLoader
 from tlz import merge, partition_all
 
 from dask.utils import format_bytes, format_time
@@ -37,7 +38,6 @@
 
 logger = logging.getLogger(__name__)
 
-from jinja2 import Environment, FileSystemLoader
 
 env = Environment(
     loader=FileSystemLoader(

distributed/dashboard/core.py

@@ -7,12 +7,13 @@
 from bokeh.application.handlers.function import FunctionHandler
 from bokeh.server.server import BokehTornado
 
+import dask
+
 try:
     from bokeh.server.util import create_hosts_allowlist
 except ImportError:
     from bokeh.server.util import create_hosts_whitelist as create_hosts_allowlist
 
-import dask
 
 if LooseVersion(bokeh.__version__) < LooseVersion("1.0.0"):
     warnings.warn(

distributed/dashboard/tests/test_components.py

@@ -2,7 +2,9 @@
 
 import pytest
 
-pytest.importorskip("bokeh")
+# isort: off
+bokeh = pytest.importorskip("bokeh")
+# isort: on
 
 from bokeh.models import ColumnDataSource, Model
 

distributed/dashboard/tests/test_scheduler_bokeh.py

@@ -7,7 +7,10 @@
 
 import pytest
 
+# isort: off
 pytest.importorskip("bokeh")
+# isort: on
+
 from bokeh.server.server import BokehTornado
 from tlz import first
 from tornado.httpclient import AsyncHTTPClient, HTTPRequest

distributed/dashboard/tests/test_worker_bokeh.py

@@ -5,7 +5,9 @@
 
 import pytest
 
+# isort: off
 pytest.importorskip("bokeh")
+# isort: on
 from tlz import first
 from tornado.httpclient import AsyncHTTPClient
 

distributed/deploy/old_ssh.py

@@ -4,16 +4,16 @@
 import sys
 import time
 import traceback
+from threading import Thread
+
+from tlz import merge
+from tornado import gen
 
 try:
     from queue import Queue
 except ImportError:  # Python 2.7 fix
     from Queue import Queue
 
-from threading import Thread
-
-from tlz import merge
-from tornado import gen
 
 logger = logging.getLogger(__name__)
 

distributed/deploy/tests/test_old_ssh.py

@@ -2,12 +2,12 @@
 
 import pytest
 
-pytest.importorskip("paramiko")
-
 from distributed import Client
 from distributed.deploy.old_ssh import SSHCluster
 from distributed.metrics import time
 
+pytest.importorskip("paramiko")
+
 
 @pytest.mark.avoid_ci
 def test_cluster(loop):

distributed/deploy/tests/test_ssh.py

@@ -1,15 +1,16 @@
-import pytest
-
-pytest.importorskip("asyncssh")
-
 import sys
 
+import pytest
+
 import dask
 
 from distributed import Client
 from distributed.compatibility import MACOS, WINDOWS
 from distributed.deploy.ssh import SSHCluster
 
+pytest.importorskip("asyncssh")
+
+
 pytestmark = [
     pytest.mark.xfail(MACOS, reason="very high flakiness; see distributed/issues/4543"),
     pytest.mark.skipif(WINDOWS, reason="no CI support; see distributed/issues/4509"),

distributed/diagnostics/tests/test_nvml.py

@@ -2,15 +2,15 @@
 
 import pytest
 
-pytestmark = pytest.mark.gpu
-
-pynvml = pytest.importorskip("pynvml")
-
 import dask
 
 from distributed.diagnostics import nvml
 from distributed.utils_test import gen_cluster
 
+pytestmark = pytest.mark.gpu
+
+pynvml = pytest.importorskip("pynvml")
+
 
 def test_one_time():
     if nvml.device_get_count() < 1:

distributed/diagnostics/tests/test_progress_stream.py

@@ -1,13 +1,13 @@
 import pytest
 
-pytest.importorskip("bokeh")
-
 from dask import delayed
 
 from distributed.client import wait
 from distributed.diagnostics.progress_stream import progress_quads, progress_stream
 from distributed.utils_test import div, gen_cluster, inc
 
+pytest.importorskip("bokeh")
+
 
 def test_progress_quads():
     msg = {