Centralize and type no_default (#8171)

distributed/client.py

@@ -35,6 +35,7 @@
 from dask.core import flatten, validate_key
 from dask.highlevelgraph import HighLevelGraph
 from dask.optimization import SubgraphCallable
+from dask.typing import no_default
 from dask.utils import (
     apply,
     ensure_dict,
@@ -101,7 +102,6 @@
     import_term,
     is_python_shutting_down,
     log_errors,
-    no_default,
     sync,
     thread_state,
 )
@@ -854,7 +854,7 @@ def __init__(
         connection_limit=512,
         **kwargs,
     ):
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = dask.config.get("distributed.comm.timeouts.connect")
         if timeout is not None:
             timeout = parse_timedelta(timeout, "s")
@@ -1248,7 +1248,7 @@ async def _start(self, timeout=no_default, **kwargs):
 
         await self.rpc.start()
 
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = self._timeout
         if timeout is not None:
             timeout = parse_timedelta(timeout, "s")
@@ -1753,7 +1753,7 @@ def close(self, timeout=no_default):
         --------
         Client.restart
         """
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = self._timeout * 2
         # XXX handling of self.status here is not thread-safe
         if self.status in ["closed", "newly-created"]:
@@ -2399,7 +2399,7 @@ async def _scatter(
         timeout=no_default,
         hash=True,
     ):
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = self._timeout
         if isinstance(workers, (str, Number)):
             workers = [workers]
@@ -2588,7 +2588,7 @@ def scatter(
         --------
         Client.gather : Gather data back to local process
         """
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = self._timeout
         if isinstance(data, pyQueue) or isinstance(data, Iterator):
             raise TypeError(
@@ -3577,7 +3577,7 @@ def persist(
             return result
 
     async def _restart(self, timeout=no_default, wait_for_workers=True):
-        if timeout == no_default:
+        if timeout is no_default:
             timeout = self._timeout * 4
         if timeout is not None:
             timeout = parse_timedelta(timeout, "s")

distributed/deploy/cluster.py

@@ -33,9 +33,6 @@
 logger = logging.getLogger(__name__)
 
 
-no_default = "__no_default__"
-
-
 class Cluster(SyncMethodMixin):
     """Superclass for cluster objects
 

distributed/scheduler.py

@@ -54,6 +54,7 @@
 import dask
 import dask.utils
 from dask.core import get_deps, validate_key
+from dask.typing import no_default
 from dask.utils import (
     format_bytes,
     format_time,
@@ -119,7 +120,6 @@
     get_fileno_limit,
     key_split_group,
     log_errors,
-    no_default,
     offload,
     recursive_to_dict,
     wait_for,
@@ -7419,7 +7419,7 @@ def get_metadata(self, keys: list[str], default: Any = no_default) -> Any:
                 metadata = metadata[key]
             return metadata
         except KeyError:
-            if default != no_default:
+            if default is not no_default:
                 return default
             else:
                 raise

distributed/tests/test_utils.py

@@ -700,6 +700,12 @@ async def set_var(v: str) -> None:
     await asyncio.gather(set_var("foo"), set_var("bar"))
 
 
+def test_no_default_deprecated():
+    with pytest.warns(FutureWarning, match="no_default is deprecated"):
+        from distributed.utils import no_default
+    assert no_default is dask.typing.no_default
+
+
 def test_iscoroutinefunction_unhashable_input():
     # Ensure iscoroutinefunction can handle unhashable callables
     assert not iscoroutinefunction(_UnhashableCallable())

distributed/utils.py

@@ -95,8 +95,6 @@
     P = ParamSpec("P")
     T = TypeVar("T")
 
-no_default = "__no_default__"
-
 _forkserver_preload_set = False
 
 
@@ -1604,7 +1602,9 @@ def clean_dashboard_address(addrs: AnyType, default_listen_ip: str = "") -> list
     return addresses
 
 
-_deprecations: dict[str, str] = {}
+_deprecations = {
+    "no_default": "dask.typing.no_default",
+}
 
 
 def __getattr__(name):