feat(core): add graph workflow group and remove subcommand (#2177)

* add graph workflow group and remove subcommand
* raise exception when provided workflow is not found
* Apply suggestions from code review
* check user provided workflow name uniqueness before executing a run
Co-authored-by: Ralf Grubenmann <[email protected]>

Author: vigsterkr

renku/cli/graph.py

@@ -22,7 +22,14 @@
 from renku.cli.utils.callback import ClickCallback
 from renku.cli.utils.click import CaseInsensitiveChoice
 from renku.core.incubation.command import Command
-from renku.core.incubation.graph import FORMATS, add_to_dataset, create_dataset, export_graph, generate_graph
+from renku.core.incubation.graph import (
+    FORMATS,
+    add_to_dataset,
+    create_dataset,
+    export_graph,
+    generate_graph,
+    remove_workflow,
+)
 from renku.core.incubation.graph import status as get_status
 from renku.core.incubation.graph import update as perform_update
 from renku.core.models.workflow.dependency_graph import DependencyGraph
@@ -189,3 +196,17 @@ def add(name, urls, external, force, overwrite, create, sources, destination, re
         ref=ref,
     )
     click.secho("OK", fg="green")
+
+
+@graph.group()
+def workflow():
+    """Proof-of-Concept command for workflow operations using new metadata."""
+
+
+@workflow.command()
+@click.argument("name", metavar="<name or uuid>")
+@click.option("-f", "--force", is_flag=True, help="Force remove (don't prompt user to confirm).")
+def remove(name, force):
+    """Remove the workflow named <name>."""
+    communicator = ClickCallback()
+    remove_workflow().with_communicator(communicator).build().execute(name=name, force=force)

renku/core/commands/run.py

@@ -25,8 +25,10 @@
 
 from renku.core import errors
 from renku.core.incubation.command import Command
+from renku.core.incubation.graph import unique_workflow
 from renku.core.management.git import get_mapped_std_streams
 from renku.core.models.cwl.command_line_tool import CommandLineToolFactory
+from renku.core.models.provenance.provenance_graph import ProvenanceGraph
 from renku.core.utils import communication
 from renku.core.utils.urls import get_slug
 
@@ -55,6 +57,12 @@ def _run_command(
         if name != valid_name:
             raise errors.ParameterError(f"Invalid name: '{name}' (Hint: '{valid_name}' is valid).")
 
+        # TODO: refactor this once we switch to Database
+        if client.provenance_graph_path.exists():
+            workflows = unique_workflow(ProvenanceGraph.from_json(client.provenance_graph_path))
+            if name in workflows:
+                raise errors.ParameterError(f"Duplicate workflow name: workflow '{name}' already exists.")
+
     paths = explicit_outputs if no_output_detection else client.candidate_paths
     mapped_std = get_mapped_std_streams(paths, streams=("stdout", "stderr"))
 

renku/core/incubation/graph.py

@@ -22,8 +22,10 @@
 import sys
 import traceback
 from collections import defaultdict
+from datetime import datetime
 from pathlib import Path
 from typing import Dict
+from urllib.parse import urlparse
 
 import git
 from git import NULL_TREE, Commit, GitCommandError
@@ -42,6 +44,8 @@
 from renku.core.models.provenance.activities import Activity
 from renku.core.models.provenance.activity import ActivityCollection
 from renku.core.models.provenance.provenance_graph import ProvenanceGraph
+from renku.core.models.workflow.dependency_graph import DependencyGraph
+from renku.core.models.workflow.plan import Plan
 from renku.core.models.workflow.run import Run
 from renku.core.utils import communication
 from renku.core.utils.contexts import measure
@@ -453,3 +457,49 @@ def _add_to_dataset(
     except (FileNotFoundError, git.exc.NoSuchPathError) as e:
         message = "\n\t".join(urls)
         raise errors.ParameterError(f"Could not find paths/URLs: \n\t{message}") from e
+
+
+def remove_workflow():
+    """Return a command for removing workflow."""
+    command = Command().command(_remove_workflow).lock_project()
+    return command.require_migration().with_commit(commit_only=GRAPH_METADATA_PATHS)
+
+
+def _remove_workflow(client, name: str, force: bool):
+    """Remove the given workflow."""
+    now = datetime.utcnow()
+    # TODO: refactor this once we switch to Database
+    provenance_graph = ProvenanceGraph.from_json(client.provenance_graph_path)
+    pg_workflows = unique_workflow(provenance_graph)
+
+    not_found_text = f'The specified workflow is "{name}" is not an active workflow.'
+    plan = None
+    parse_result = urlparse(name)
+    if parse_result.scheme:
+        plan = next(filter(lambda x: x.id == name, pg_workflows.values()), None)
+    if not plan and name not in pg_workflows:
+        raise errors.ParameterError(not_found_text)
+
+    if not force:
+        prompt_text = f'You are about to remove the following workflow "{name}".' + "\n" + "\nDo you wish to continue?"
+        communication.confirm(prompt_text, abort=True, warning=True)
+
+    plan = plan or pg_workflows[name]
+    plan.invalidated_at = now
+    dependency_graph = DependencyGraph.from_json(client.dependency_graph_path)
+    for p in dependency_graph.plans:
+        if p.id == plan.id:
+            p.invalidated_at = now
+
+    dependency_graph.to_json()
+    provenance_graph.to_json()
+
+
+def unique_workflow(provenance_graph: ProvenanceGraph) -> Dict[str, Plan]:
+    """Map of unique plans in the provenance graph indexed by name."""
+    workflows = dict()
+    for activity in provenance_graph.activities:
+        plan = activity.association.plan
+        if plan.invalidated_at is None and plan.name not in workflows:
+            workflows[plan.name] = plan
+    return workflows

renku/core/models/workflow/plan.py

@@ -19,6 +19,7 @@
 
 import copy
 import itertools
+from datetime import datetime
 from pathlib import PurePosixPath
 from typing import Any, Dict, List
 from uuid import uuid4
@@ -54,6 +55,7 @@ def __init__(
         description: str = None,
         id: str,
         inputs: List[CommandInput] = None,
+        invalidated_at: datetime = None,
         keywords: List[str] = None,
         name: str = None,
         outputs: List[CommandOutput] = None,
@@ -63,6 +65,7 @@ def __init__(
         self.description: str = description
         self.id: str = id
         self.inputs: List[CommandInput] = inputs or []
+        self.invalidated_at: datetime = invalidated_at
         self.keywords: List[str] = keywords or []
         self.name: str = name
         self.outputs: List[CommandOutput] = outputs or []
@@ -281,6 +284,7 @@ class Meta:
     description = fields.String(schema.description, missing=None)
     id = fields.Id()
     inputs = Nested(renku.hasInputs, CommandInputSchema, many=True, missing=None)
+    invalidated_at = fields.DateTime(prov.invalidatedAtTime, add_value_types=True)
     keywords = fields.List(schema.keywords, fields.String(), missing=None)
     name = fields.String(schema.name, missing=None)
     outputs = Nested(renku.hasOutputs, CommandOutputSchema, many=True, missing=None)

tests/core/incubation/test_workflow.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2017-2021- Swiss Data Science Center (SDSC)
+# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+# Eidgenössische Technische Hochschule Zürich (ETHZ).
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Test workflow commands."""
+
+# from renku.core.incubation.command import Command, CommandResult
+from renku.cli import cli
+
+
+def test_workflow_remove_command(runner, project):
+    """test workflow remove with builder."""
+    # TODO: add fixture project with graph generated by default
+    result = runner.invoke(cli, ["graph", "generate"])
+    assert 0 == result.exit_code
+
+    workflow_name = "test_workflow"
+
+    result = runner.invoke(cli, ["graph", "workflow", "remove", workflow_name])
+    assert 2 == result.exit_code
+
+    result = runner.invoke(cli, ["run", "--success-code", "0", "--no-output", "--name", workflow_name, "echo", "foo"])
+    assert 0 == result.exit_code
+
+    result = runner.invoke(cli, ["graph", "workflow", "remove", "--force", workflow_name])
+    assert 0 == result.exit_code