72 support non editable workflows 2

CHANGELOG.md

@@ -20,7 +20,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - New API version `v2_1_0`:
-- New endpoint `POST /api/workflows/discover` to discover ewoks workflows from python packages.
+- New endpoint `POST /api/workflows/discover` to discover ewoks workflows from python projects
+  with a local copy that shadows the remote content.
+  - Local shadowing on discovery can be disabled.
+  - Editing a remote workflow creates a local shadow.
+  - Deleting a remote workflow without a local shadow fails.
+  - Re-discovering does not override shadows.
 
 ## [2.1.2] - 2026-03-06
 

src/ewoksserver/app/lifespan.py

@@ -17,6 +17,7 @@
 from .backends import json_backend
 from .routes.common import discovery
 from .routes.execution import socketio
+from .routes.workflows import backend as workflow_backend
 
 logger = logging.getLogger(__name__)
 
@@ -78,10 +79,14 @@ def _rediscover_resources(ewoks_settings: config.EwoksSettings) -> None:
         json_backend.save_resource(root_url, resource["task_identifier"], resource)
 
     try:
-        discovery.discover_workflows(ewoks_settings)
+        _, identifier_to_queue = discovery.discover_workflows(ewoks_settings)
     except Exception as ex:
+        identifier_to_queue = {}
         logger.exception("Workflow discovery failed: %s", ex)
-    logger.warning("Discovered workflows not used yet")
+    root_url = json_backend.root_url(ewoks_settings.resource_directory, "workflows")
+    workflow_backend.register_remote_workflows(
+        ewoks_settings, root_url, identifier_to_queue
+    )
 
 
 def _enable_execution_events(ewoks_settings: config.EwoksSettings) -> None:

src/ewoksserver/app/models.py

@@ -15,9 +15,15 @@ class EwoksSchedulingType(str, Enum):
 
 
 class EwoksDiscoverySettings(BaseModel):
-    on_start_up: bool = Field(default=True, title="Discover ewoks tasks on startup")
+    on_start_up: bool = Field(
+        default=True, title="Discover ewoks tasks/workflows on startup"
+    )
     timeout: float | None = Field(
-        default=None, title="Timeout for task discovery (in seconds)"
+        default=None, title="Timeout for task/workflow discovery (in seconds)"
+    )
+    cache_workflows: bool = Field(
+        default=True,
+        title="Create a local copy of a workflow when it is discovered",
     )
 
 

src/ewoksserver/app/routes/common/discovery.py

@@ -51,7 +51,7 @@ def discover_tasks(
     if task_type is not None:
         discover_kwargs["task_type"] = task_type
 
-    tasks = _discover(
+    tasks, _identifier_to_queue = _discover(
         discover,
         settings,
         modules=modules,
@@ -70,11 +70,14 @@ def discover_workflows(
     modules: list[str] | None = None,
     workflow_extension: str | None = None,
     worker_options: dict | None = None,
-) -> list[str]:
+) -> tuple[list[str], dict[str, str | None]]:
     """
     :raises ModuleNotFoundError: failed importing workflows.
     :raises TimeoutError: timeout when asking a remote worker for workflows.
     :raises Exception: any other import or remote error.
+    :returns: the discovered workflow identifiers, and a mapping of each
+        identifier to the celery queue it was discovered on (`None` for
+        local scheduling).
     """
     if settings.ewoks_scheduling.type == EwoksSchedulingType.Local:
         if modules:
@@ -108,11 +111,14 @@ def _discover(
     discover_kwargs: dict,
     worker_options: dict | None,
     id_extractor: Callable[[Any], str],
-) -> list:
+) -> tuple[list, dict[str, str | None]]:
     """
     :raises ModuleNotFoundError: failed importing tasks or workflows.
     :raises TimeoutError: timeout when asking a remote worker.
     :raises Exception: any other import or remote error.
+    :returns: the discovered items, and a mapping of each item's identifier
+        (see `id_extractor`) to the celery queue it was discovered on
+        (`None` for local scheduling).
     """
     if worker_options is None:
         kwargs = dict()
@@ -128,7 +134,8 @@ def _discover(
 
     timeout = settings.ewoks_discovery.timeout
     if settings.ewoks_scheduling.type == EwoksSchedulingType.Local:
-        return _discover_locally(discover, kwargs, timeout=timeout)
+        items = _discover_locally(discover, kwargs, timeout=timeout)
+        return items, {id_extractor(item): None for item in items}
     else:
         return _discover_in_all_queues(discover, kwargs, id_extractor, timeout=timeout)
 
@@ -142,12 +149,13 @@ def _discover_in_all_queues(
     kwargs: dict,
     id_extractor: Callable[[Any], str],
     timeout: float | None = None,
-) -> list:
-    futures = [discover(**kwargs, queue=queue) for queue in get_queues()]
+) -> tuple[list, dict[str, str | None]]:
+    futures = [(queue, discover(**kwargs, queue=queue)) for queue in get_queues()]
 
     # Store items in a dict to avoid duplicates
     item_dict = {}
-    for future in futures:
+    identifier_to_queue: dict[str, str | None] = {}
+    for queue, future in futures:
         # Ignore failures of a single queue to not prevent discovery on other queues
         new_items = future.result(timeout=timeout)
         exc = future.exception()
@@ -157,8 +165,10 @@ def _discover_in_all_queues(
         if new_items is None:
             continue
         for item in new_items:
-            item_dict[id_extractor(item)] = item
-    return list(item_dict.values())
+            identifier = id_extractor(item)
+            item_dict[identifier] = item
+            identifier_to_queue.setdefault(identifier, queue)
+    return list(item_dict.values()), identifier_to_queue
 
 
 def _set_default_task_properties(task: dict) -> None:

src/ewoksserver/app/routes/workflows/backend.py

@@ -0,0 +1,222 @@
+import json
+import logging
+from pathlib import Path
+from typing import Any
+from typing import Iterator
+
+from ewoksjob.client import convert_graph
+from ewoksjob.client.local import convert_graph as convert_graph_local
+
+from ...backends import json_backend
+from ...config import EwoksSettings
+from ...models import EwoksSchedulingType
+
+logger = logging.getLogger(__name__)
+
+
+def load_workflow(
+    settings: EwoksSettings,
+    root: json_backend.ResourceUrlType,
+    identifier: str,
+    worker_options: dict | None = None,
+) -> json_backend.ResourceContentType:
+    """Load a local and remote workflow.
+
+    :raises FileNotFoundError: no local and remote workflow
+        for this identifier.
+    """
+    if json_backend.resource_exists(root, identifier):
+        return json_backend.load_resource(root, identifier)
+
+    index = _load_remote_workflow_index(settings)
+    if identifier not in index:
+        raise FileNotFoundError(identifier)
+
+    graph = _load_remote_workflow(
+        settings, identifier, queue=index[identifier], worker_options=worker_options
+    )
+    if graph is None:
+        raise FileNotFoundError(identifier)
+    graph.setdefault("graph", {})["id"] = identifier
+    return graph
+
+
+def save_workflow(
+    settings: EwoksSettings,
+    root: json_backend.ResourceUrlType,
+    identifier: str,
+    content: json_backend.ResourceContentType,
+) -> None:
+    """Save a workflow, turning it from a remote into a local shadowing
+    workflow if it was still remote.
+
+    :raises PermissionError: no permission to save the workflow.
+    """
+    _shadow_if_remote_workflow(settings, identifier)
+    json_backend.save_resource(root, identifier, content)
+
+
+def delete_workflow(root: json_backend.ResourceUrlType, identifier: str) -> None:
+    """Delete a local shadowing workflow.
+
+    :raises PermissionError: no permission to delete the workflow.
+    :raises FileNotFoundError: no local workflow for this identifier.
+    """
+    json_backend.delete_resource(root, identifier)
+
+
+def workflow_exists(
+    settings: EwoksSettings, root: json_backend.ResourceUrlType, identifier: str
+) -> bool:
+    """Whether a local or remote workflow exists for this identifier.
+
+    :raises ValueError: invalid identifier.
+    """
+    return json_backend.resource_exists(root, identifier) or is_remote_workflow(
+        settings, identifier
+    )
+
+
+def workflow_identifiers(
+    settings: EwoksSettings, root: json_backend.ResourceUrlType
+) -> list[str]:
+    """Identifiers of local and remote workflows."""
+    identifiers = set(json_backend.resource_identifiers(root))
+    identifiers.update(_load_remote_workflow_index(settings))
+    return sorted(identifiers)
+
+
+def iter_workflow_graphs(
+    settings: EwoksSettings,
+    root: json_backend.ResourceUrlType,
+    worker_options: dict | None = None,
+) -> Iterator[dict]:
+    """Yield `graph` attributes of local or remote workflows."""
+    shadowed = set()
+    for identifier in json_backend.resource_identifiers(root):
+        shadowed.add(identifier)
+        yield json_backend.load_resource(root, identifier).get("graph", {})
+
+    index = _load_remote_workflow_index(settings)
+    for identifier, queue in index.items():
+        if identifier in shadowed:
+            continue
+        graph = _load_remote_workflow(
+            settings, identifier, queue=queue, worker_options=worker_options
+        )
+        if graph is None:
+            continue
+        graph.setdefault("graph", {})["id"] = identifier
+        yield graph["graph"]
+
+
+def register_remote_workflows(
+    settings: EwoksSettings,
+    root: json_backend.ResourceUrlType,
+    identifier_to_queue: dict[str, str | None],
+    worker_options: dict | None = None,
+) -> None:
+    """Register discovered remote workflows, skipping ones already
+    shadowed locally. Persists a shadow right away if `cache_workflows`
+    is enabled.
+    """
+    create_shadow = settings.ewoks_discovery.cache_workflows
+    index = _load_remote_workflow_index(settings)
+    index_changed = False
+    for identifier, queue in identifier_to_queue.items():
+        if json_backend.resource_exists(root, identifier):
+            continue
+
+        if create_shadow:
+            graph = _load_remote_workflow(
+                settings, identifier, queue=queue, worker_options=worker_options
+            )
+            if graph is None:
+                continue
+            graph.setdefault("graph", {})["id"] = identifier
+            json_backend.save_resource(root, identifier, graph)
+
+        if identifier not in index or index[identifier] != queue:
+            index[identifier] = queue
+            index_changed = True
+
+    if index_changed:
+        _save_remote_workflow_index(settings, index)
+
+
+def is_remote_workflow(settings: EwoksSettings, identifier: str) -> bool:
+    """Whether a workflow is registered as a remote workflow."""
+    return identifier in _load_remote_workflow_index(settings)
+
+
+_REMOTE_WORKFLOW_INDEX = "remote_workflow_index.json"
+
+
+def _remote_workflow_index_path(settings: EwoksSettings) -> Path:
+    return settings.resource_directory / _REMOTE_WORKFLOW_INDEX
+
+
+def _load_remote_workflow_index(settings: EwoksSettings) -> dict[str, Any]:
+    """The remote workflow index: identifier -> discovery queue."""
+    try:
+        with open(_remote_workflow_index_path(settings)) as f:
+            return json.load(f)
+    except FileNotFoundError:
+        return {}
+
+
+def _save_remote_workflow_index(settings: EwoksSettings, index: dict[str, Any]) -> None:
+    """The remote workflow index: identifier -> discovery queue."""
+    path = _remote_workflow_index_path(settings)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w") as f:
+        json.dump(index, f, indent=2)
+
+
+def _shadow_if_remote_workflow(settings: EwoksSettings, identifier: str) -> None:
+    """Turn a remote workflow into a local shadowing workflow.
+    The local copy is expected to be created by the caller."""
+    index = _load_remote_workflow_index(settings)
+    if identifier in index:
+        del index[identifier]
+        _save_remote_workflow_index(settings, index)
+
+
+def _load_remote_workflow(
+    settings: EwoksSettings,
+    identifier: str,
+    queue: str | None = None,
+    worker_options: dict | None = None,
+) -> dict | None:
+    """Load a remote workflow identified by its fully qualified
+    module identifier, e.g.``"mypackage.subpackage.myworkflow"``.
+
+    :returns: `None` when the workflow could not be loaded.
+    """
+    package, _, _ = identifier.rpartition(".")
+    if not package:
+        return None
+
+    if worker_options is None:
+        kwargs = dict()
+    else:
+        kwargs = dict(worker_options)
+    kwargs["args"] = (identifier, None)
+    kwargs["kwargs"] = {
+        "load_options": {"representation": "json_module", "root_module": package}
+    }
+
+    timeout = settings.ewoks_discovery.timeout
+    try:
+        if settings.ewoks_scheduling.type == EwoksSchedulingType.Local:
+            future = convert_graph_local(**kwargs)
+        else:
+            future = convert_graph(**kwargs, queue=queue)
+        graph = future.result(timeout=timeout)
+    except Exception as ex:
+        logger.warning("Failed to load remote workflow %r: %s", identifier, ex)
+        return None
+
+    if not isinstance(graph, dict):
+        return None
+    return graph

src/ewoksserver/app/routes/workflows/descriptions.py

@@ -1,6 +1,8 @@
 from typing import Iterator
 
 from ...backends import json_backend
+from ...config import EwoksSettings
+from . import backend
 
 _WORKFLOW_KEYWORDS = (
     "id",
@@ -13,10 +15,11 @@
 
 
 def workflow_descriptions(
-    root: json_backend.ResourceUrlType, keywords: dict | None = None
+    settings: EwoksSettings,
+    root: json_backend.ResourceUrlType,
+    keywords: dict | None = None,
 ) -> Iterator[dict]:
-    for res in json_backend.resources(root):
-        description = res["graph"]
+    for description in backend.iter_workflow_graphs(settings, root):
         if not _include_resource(description.get("keywords", dict()), keywords):
             continue
         yield {