Make dask[distributed] optional via covalent[dask]- no dask install by default

.github/workflows/benchmark.yml

@@ -59,7 +59,7 @@ jobs:
         run: python setup.py sdist && pip install dist/*.tar.gz
 
       - name: Install test requirements
-        run: pip install --no-cache-dir -r ./tests/requirements.txt
+        run: pip install --no-cache-dir -r ./tests/requirements.txt "locust>=2.11.0"
 
       - name: Start Covalent dispatcher server
         run: covalent start -d --ignore-migrations

.github/workflows/release.yml

@@ -164,7 +164,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v2
         with:
-          python-version: "3.10"
+          python-version: "3.13"
 
       - name: Install Python dependencies
         run: |

.github/workflows/requirements.yml

@@ -30,7 +30,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v2
         with:
-          python-version: "3.10"
+          python-version: "3.13"
 
       - name: Install covalent
         run: python -m pip install .

.github/workflows/test_matrix.json

@@ -1,20 +1,4 @@
 [
-  {
-    "name": "Debian 12 / Python 3.10 / Dask",
-    "os": "ubuntu-latest",
-    "container": "ghcr.io/agnostiqhq/covalent-dev/debian12-py310:latest",
-    "backend": "dask",
-    "experimental": false,
-    "trigger": ["schedule", "workflow_dispatch", "pull_request"]
-  },
-  {
-    "name": "Debian 12 / Python 3.10 / Local",
-    "os": "ubuntu-latest",
-    "container": "ghcr.io/agnostiqhq/covalent-dev/debian12-py310:latest",
-    "backend": "local",
-    "experimental": false,
-    "trigger": ["schedule", "workflow_dispatch", "pull_request"]
-  },
   {
     "name": "Debian 12 / Python 3.11 / Dask",
     "os": "ubuntu-latest",
@@ -48,20 +32,20 @@
     "trigger": ["schedule", "workflow_dispatch", "pull_request"]
   },
   {
-    "name": "MacOS 13 / Python 3.10 / Dask",
-    "os": "macos-13",
-    "python-version": "3.10",
+    "name": "Debian 13 / Python 3.13 / Dask",
+    "os": "ubuntu-latest",
+    "container": "ghcr.io/agnostiqhq/covalent-dev/debian13-py313:latest",
     "backend": "dask",
     "experimental": false,
-    "trigger": ["schedule", "workflow_dispatch"]
+    "trigger": ["push", "schedule", "workflow_dispatch", "pull_request"]
   },
   {
-    "name": "MacOS 13 / Python 3.10 / Local",
-    "os": "macos-13",
-    "python-version": "3.10",
+    "name": "Debian 13 / Python 3.13 / Local",
+    "os": "ubuntu-latest",
+    "container": "ghcr.io/agnostiqhq/covalent-dev/debian13-py313:latest",
     "backend": "local",
     "experimental": false,
-    "trigger": ["schedule", "workflow_dispatch"]
+    "trigger": ["schedule", "workflow_dispatch", "pull_request"]
   },
   {
     "name": "MacOS 13 / Python 3.11 / Dask",
@@ -88,25 +72,25 @@
     "trigger": ["schedule", "workflow_dispatch"]
   },
   {
-    "name": "MacOS 13 / Python 3.11 / Local",
+    "name": "MacOS 13 / Python 3.12 / Local",
     "os": "macos-13",
-    "python-version": "3.11",
+    "python-version": "3.12",
     "backend": "local",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
   },
   {
-    "name": "MacOS 14 / Python 3.10 / Dask",
-    "os": "macos-14",
-    "python-version": "3.10",
+    "name": "MacOS 13 / Python 3.13 / Dask",
+    "os": "macos-13",
+    "python-version": "3.13",
     "backend": "dask",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
   },
   {
-    "name": "MacOS 14 / Python 3.10 / Local",
-    "os": "macos-14",
-    "python-version": "3.10",
+    "name": "MacOS 13 / Python 3.13 / Local",
+    "os": "macos-13",
+    "python-version": "3.13",
     "backend": "local",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
@@ -144,17 +128,17 @@
     "trigger": ["schedule", "workflow_dispatch"]
   },
   {
-    "name": "MacOS 15 / Python 3.10 / Dask",
-    "os": "macos-15",
-    "python-version": "3.10",
+    "name": "MacOS 14 / Python 3.13 / Dask",
+    "os": "macos-14",
+    "python-version": "3.13",
     "backend": "dask",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
   },
   {
-    "name": "MacOS 15 / Python 3.10 / Local",
-    "os": "macos-15",
-    "python-version": "3.10",
+    "name": "MacOS 14 / Python 3.13 / Local",
+    "os": "macos-14",
+    "python-version": "3.13",
     "backend": "local",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
@@ -190,5 +174,21 @@
     "backend": "local",
     "experimental": false,
     "trigger": ["schedule", "workflow_dispatch"]
+  },
+  {
+    "name": "MacOS 15 / Python 3.13 / Dask",
+    "os": "macos-15",
+    "python-version": "3.13",
+    "backend": "dask",
+    "experimental": false,
+    "trigger": ["schedule", "workflow_dispatch"]
+  },
+  {
+    "name": "MacOS 15 / Python 3.13 / Local",
+    "os": "macos-15",
+    "python-version": "3.13",
+    "backend": "local",
+    "experimental": false,
+    "trigger": ["schedule", "workflow_dispatch"]
   }
 ]

.github/workflows/tests.yml

@@ -184,7 +184,9 @@ jobs:
       - name: Build distribution
         id: build-dist
         if: env.NEED_PYTHON || env.BUILD_AND_RUN_ALL
-        run: python setup.py sdist
+        run: |
+          pip install --no-cache-dir setuptools
+          python setup.py sdist
 
       - name: Validate distribution
         if: env.BUILD_AND_RUN_ALL

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [UNRELEASED]
 
+### Changed
+
+- Make `dask[distributed]` and optional dependency via `covalent[dask]` extra
+- Default executor changed from `dask` to `local` in `defaults.py`
+- Dropped Python 3.10 support, added Python 3.13; updated `test_matrix.json` and CI workflows accordingly
+- Fixed `extract_graph` to handle both `"edges"` (networkx ≥3.x) and `"links"` (older networkx) as the edge key; added lower bound on networkx version
+- Fixed various 422 errors by setting additional `Content-Type: application/json` headers as necessary
+- Moved `locust` out of `tests/requirements.txt` into the benchmark workflow only, to avoid a `gevent` conflict
+- Various test fixes: renamed `status` → `job_status`, corrected mock paths, updated graph key references
+
 ## [0.240.0-rc.0] - 2025-05-14
 
 ### Authors

covalent/_dispatcher_plugins/local.py

@@ -482,7 +482,11 @@ def register_manifest(
         stripped = strip_local_uris(manifest) if push_assets else manifest
         endpoint = BASE_ENDPOINT
 
-        r = APIClient(dispatcher_addr).post(endpoint, data=stripped.model_dump_json())
+        r = APIClient(dispatcher_addr).post(
+            endpoint,
+            data=stripped.model_dump_json(),
+            headers={"Content-Type": "application/json"},
+        )
         r.raise_for_status()
 
         parsed_resp = ResultSchema.model_validate(r.json())
@@ -513,7 +517,10 @@ def register_derived_manifest(
 
         params = {"reuse_previous_results": reuse_previous_results}
         r = APIClient(dispatcher_addr).post(
-            endpoint, data=stripped.model_dump_json(), params=params
+            endpoint,
+            data=stripped.model_dump_json(),
+            headers={"Content-Type": "application/json"},
+            params=params,
         )
         r.raise_for_status()
 

covalent/_serialize/transport_graph.py

@@ -132,7 +132,7 @@ def serialize_transport_graph(tg, storage_path: str) -> TransportGraphSchema:
     g = tg.get_internal_graph_copy()
     return TransportGraphSchema(
         nodes=_serialize_nodes(g, storage_path),
-        links=_serialize_edges(g),
+        edges=_serialize_edges(g),
     )
 
 
@@ -151,7 +151,7 @@ def deserialize_transport_graph(t: TransportGraphSchema) -> _TransportGraph:
     tg = _TransportGraph()
     g = tg._graph
     nodes = [deserialize_node(n) for n in t.nodes]
-    edges = [_deserialize_edge(e) for e in t.links]
+    edges = [_deserialize_edge(e) for e in t.edges]
     for node in nodes:
         node_id = node["id"]
         attrs = node["attrs"]

covalent/_shared_files/defaults.py

@@ -20,7 +20,14 @@
 from dataclasses import dataclass, field
 from typing import Dict
 
-import dask.system
+try:
+    import dask.system
+
+    _dask_cpu_count = dask.system.CPU_COUNT
+    _DASK_AVAILABLE = True
+except ImportError:
+    _dask_cpu_count = os.cpu_count() or 1
+    _DASK_AVAILABLE = False
 
 prefix_separator = ":"
 
@@ -129,7 +136,7 @@ def get_default_dask_config():
         ),
         "mem_per_worker": "auto",
         "threads_per_worker": 1,
-        "num_workers": dask.system.CPU_COUNT,
+        "num_workers": _dask_cpu_count,
     }
 
 
@@ -172,7 +179,9 @@ def get_default_executor() -> dict:
 
     return (
         "local"
-        if os.environ.get("COVALENT_DISABLE_DASK") == "1" or get_config("sdk.no_cluster") == "true"
+        if not _DASK_AVAILABLE
+        or os.environ.get("COVALENT_DISABLE_DASK") == "1"
+        or get_config("sdk.no_cluster") == "true"
         else "dask"
     )
 

covalent/_shared_files/schemas/transport_graph.py

@@ -26,7 +26,7 @@
 
 class TransportGraphSchema(BaseModel):
     nodes: List[ElectronSchema]
-    links: List[EdgeSchema]
+    edges: List[EdgeSchema]
 
     # For use by redispatch
     def reset(self):

covalent/_workflow/transport.py

@@ -411,10 +411,10 @@ def serialize(self, metadata_only: bool = False) -> bytes:
                         data["nodes"][idx].pop(field, None)
 
             # Remove the non-source-target fields from the scheduler workflow input data.
-            for idx, node in enumerate(data["links"]):
-                for name in data["links"][idx].copy():
+            for idx, node in enumerate(data["edges"]):
+                for name in data["edges"][idx].copy():
                     if name not in ["source", "target"]:
-                        data["links"][idx].pop("edge_name", None)
+                        data["edges"][idx].pop("edge_name", None)
 
         data["lattice_metadata"] = self.lattice_metadata
         return cloudpickle.dumps(data)
@@ -475,10 +475,10 @@ def serialize_to_json(self, metadata_only: bool = False) -> str:
                         data["nodes"][idx].pop(field, None)
 
             # Remove the non-source-target fields from the scheduler workflow input data.
-            for idx, node in enumerate(data["links"]):
-                for name in data["links"][idx].copy():
+            for idx, node in enumerate(data["edges"]):
+                for name in data["edges"][idx].copy():
                     if name not in ["source", "target"]:
-                        data["links"][idx].pop("edge_name", None)
+                        data["edges"][idx].pop("edge_name", None)
 
         data["lattice_metadata"] = encode_metadata(self.lattice_metadata)
         return json.dumps(data)

covalent/executor/__init__.py

@@ -242,7 +242,10 @@ def _load_executors(self, executor_dir: str) -> None:
                     # Import the module that contains the plugin
                     module_spec = importlib.util.spec_from_file_location(module_name, module_file)
                     the_module = importlib.util.module_from_spec(module_spec)
-                    module_spec.loader.exec_module(the_module)
+                    try:
+                        module_spec.loader.exec_module(the_module)
+                    except ImportError:
+                        continue
 
                     self._populate_executor_map_from_module(the_module)
 

covalent/executor/executor_plugins/dask.py

@@ -27,7 +27,10 @@
 from enum import Enum
 from typing import Any, Callable, Dict, List, Literal, Optional
 
-from dask.distributed import CancelledError, Client, Future
+try:
+    from dask.distributed import CancelledError, Client, Future
+except ImportError as e:
+    raise ImportError("'covalent[dask]' is required for DaskExecutor.") from e
 from pydantic import BaseModel
 
 from covalent._shared_files import TaskRuntimeError, logger

covalent_dispatcher/_cli/service.py

@@ -33,14 +33,9 @@
 from typing import Optional
 
 import click
-import dask.system
 import psutil
 import requests
 import sqlalchemy
-from dask.distributed import Client
-from distributed.comm import unparse_address
-from distributed.comm.core import CommClosedError
-from distributed.core import connect, rpc
 from furl import furl
 from natsort import natsorted
 from rich.box import ROUNDED
@@ -57,6 +52,19 @@
 
 from .._db.datastore import DataStore
 
+_DASK_AVAILABLE = False
+try:
+    import dask.system as _dask_system
+    from dask.distributed import Client
+    from distributed.comm import unparse_address
+    from distributed.comm.core import CommClosedError
+    from distributed.core import connect, rpc
+
+    _DASK_AVAILABLE = True
+except ImportError:
+    CommClosedError = OSError  # safe stand-in; these code paths are never reached without dask
+
+
 UI_PIDFILE = get_config("dispatcher.cache_dir") + "/ui.pid"
 UI_LOGFILE = get_config("user_interface.log_dir") + "/covalent_ui.log"
 UI_SRVDIR = f"{os.path.dirname(os.path.abspath(__file__))}/../../covalent_ui"
@@ -855,13 +863,15 @@ async def _get_cluster_logs(uri):
 
 
 def _get_cluster_admin_address():
+    if not _DASK_AVAILABLE:
+        return None
     try:
         admin_host = get_config("dask.admin_host")
         admin_port = get_config("dask.admin_port")
         admin_server_addr = unparse_address("tcp", f"{admin_host}:{admin_port}")
         return admin_server_addr
     except KeyError:
-        return
+        return None
 
 
 @click.command()
@@ -877,7 +887,7 @@ def _get_cluster_admin_address():
     is_flag=False,
     nargs=1,
     type=int,
-    default=dask.system.CPU_COUNT,
+    default=_dask_system.CPU_COUNT if _DASK_AVAILABLE else (os.cpu_count() or 1),
     show_default=True,
     help="Scale cluster by adding/removing workers to match `nworkers`",
 )