feat(api): activity (#2911)

Author: mohammad-alisafaee

docs/reference/api.rst

@@ -39,7 +39,14 @@ Renku Python API
 
 .. automodule:: renku.ui.api.models.plan
 
-.. _api-run:
+.. _api-activity:
+
+``Activity``
+------------
+
+.. automodule:: renku.ui.api.models.activity
+
+.. _api-parameter:
 
 ``Inputs, Outputs, and Parameters``
 -----------------------------------

pyproject.toml

@@ -325,11 +325,12 @@ flake8-max-line-length = 120
 testpaths = ["docs", "tests", "renku", "conftest.py"]
 markers = [
     "integration: mark a test as a integration.",
-    "service: mark a test as service test.",
     "jobs: mark a test as a job test.",
     "migration: mark a test as a migration test.",
-    "shelled: mark a test as a shelled test.",
+    "publish: mark tests that publish datasets to external providers.",
     "serial: mark a test that can not be run in parallel",
+    "service: mark a test as service test.",
+    "shelled: mark a test as a shelled test.",
 ]
 
 [tool.mypy]

renku/__init__.py

@@ -96,7 +96,7 @@ def find_spec(self, fullname, path, target=None):
         return None
 
 
-# NOTE: Patch python impoprt machinery with custom loader
+# NOTE: Patch python import machinery with custom loader
 sys.meta_path.insert(
     0,
     DeprecatedImportInterceptor(
@@ -105,7 +105,7 @@ def find_spec(self, fullname, path, target=None):
             "renku.core.metadata": ("renku.infrastructure", False),
             "renku.core.commands": ("renku.command", True),
             "renku.core.plugins": ("renku.core.plugin", True),
-            "renku.api": ("renku.ui.api", True),
+            "renku.api": ("renku.ui.api", False),
             "renku.cli": ("renku.ui.cli", True),
         }
     ),

renku/command/command_builder/command.py

@@ -102,7 +102,7 @@ def remove_injector():
 
 
 @contextlib.contextmanager
-def replace_injection(bindings, constructor_bindings=None):
+def replace_injection(bindings: Dict, constructor_bindings=None):
     """Temporarily inject various test objects.
 
     Args:

renku/core/interface/activity_gateway.py

@@ -35,6 +35,10 @@ def get_all_generation_paths(self) -> List[str]:
         """Return all generation paths."""
         raise NotImplementedError
 
+    def get_activities_by_usage(self, path: Union[Path, str], checksum: Optional[str] = None) -> List[Activity]:
+        """Return the list of all activities that use a path."""
+        raise NotImplementedError
+
     def get_activities_by_generation(self, path: Union[Path, str], checksum: Optional[str] = None) -> List[Activity]:
         """Return the list of all activities that generate a path."""
         raise NotImplementedError
@@ -43,6 +47,10 @@ def get_downstream_activities(self, activity: Activity, max_depth=None) -> Set[A
         """Get downstream activities that depend on this activity."""
         raise NotImplementedError
 
+    def get_upstream_activities(self, activity: Activity, max_depth=None) -> Set[Activity]:
+        """Get upstream activities that this activity depends on."""
+        raise NotImplementedError
+
     def get_downstream_activity_chains(self, activity: Activity) -> List[Tuple[Activity, ...]]:
         """Get a list of tuples of all downstream paths of this activity."""
         raise NotImplementedError

renku/core/migration/m_0005__2_cwl.py

@@ -195,13 +195,13 @@ def _migrate_single_step(client, cmd_line_tool, path, commit=None, parent_commit
         if matched_output:
             outputs.remove(matched_output)
 
-    created_outputs = []
+    generated_outputs = []
     workdir_requirements = [r for r in cmd_line_tool.requirements if isinstance(r, InitialWorkDirRequirement)]
 
     for r in workdir_requirements:
         for listing in r.listing:
             if listing.entry == '$({"listing": [], "class": "Directory"})':
-                created_outputs.append(listing.entryname)
+                generated_outputs.append(listing.entryname)
 
     # NOTE: multiple outputs might bind to the same input; we use this copy to find output bindings
     all_inputs = inputs.copy()
@@ -244,7 +244,7 @@ def _migrate_single_step(client, cmd_line_tool, path, commit=None, parent_commit
         if not (client.path / path).is_dir():
             check_path = path.parent
 
-        if check_path != "." and str(check_path) in created_outputs:
+        if check_path != "." and str(check_path) in generated_outputs:
             create_folder = True
 
         run.outputs.append(

renku/core/util/os.py

@@ -16,7 +16,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """OS utility functions."""
-
+import fnmatch
 import hashlib
 import os
 import re
@@ -205,3 +205,17 @@ def safe_read_yaml(file: str) -> Dict[str, Any]:
         return yaml.read_yaml(file)
     except Exception as e:
         raise errors.ParameterError(e)
+
+
+def matches(path: Union[Path, str], pattern: str) -> bool:
+    """Check if a path matched a given pattern."""
+    pattern = pattern.rstrip(os.sep)
+
+    path = Path(path)
+    paths = [path] + list(path.parents)[:-1]
+
+    for parent in paths:
+        if fnmatch.fnmatch(str(parent), pattern):
+            return True
+
+    return False

renku/core/workflow/value_resolution.py

@@ -89,7 +89,9 @@ def apply(self) -> AbstractPlan:
         params_map = TemplateVariableFormatter.to_map(chain(self._plan.inputs, self._plan.parameters))
         for param in chain(self._plan.inputs, self._plan.parameters, self._plan.outputs):
             if isinstance(param.actual_value, str):
+                actual_value_set_before = getattr(param, "_v_actual_value_set", False)
                 param.actual_value = self._template_engine.apply(param.actual_value, params_map)
+                param._v_actual_value_set = actual_value_set_before
 
         self.missing_parameters = values_keys
 

renku/domain_model/provenance/activity.py

@@ -134,6 +134,7 @@ def from_plan(
         started_at_time: datetime,
         ended_at_time: datetime,
         annotations: List[Annotation] = None,
+        id: Optional[str] = None,
         update_commits=False,
     ):
         """Convert a ``Plan`` to a ``Activity``."""
@@ -145,7 +146,7 @@ def from_plan(
         generations = {}
         parameter_values = []
 
-        activity_id = cls.generate_id()
+        activity_id = id or cls.generate_id()
 
         for input in plan.inputs:
             input_path = input.actual_value
@@ -212,6 +213,9 @@ def from_plan(
 
         return activity
 
+    def __repr__(self):
+        return f"<Activity '{self.id}': {self.association.plan.name} @ {self.ended_at_time}>"
+
     @cached_property
     def plan_with_values(self) -> Plan:
         """Get a copy of the associated plan with values from ParameterValues applied."""

renku/domain_model/provenance/parameter.py

@@ -17,11 +17,9 @@
 # limitations under the License.
 """Classes for tracking parameter values in provenance."""
 
-from itertools import chain
 from typing import Any
 from uuid import uuid4
 
-from renku.core import errors
 from renku.domain_model.workflow.plan import Plan
 
 
@@ -40,9 +38,5 @@ def generate_id(activity_id: str) -> str:
 
     def apply_value_to_parameter(self, plan: Plan) -> None:
         """Apply the current value as actual_value on the plan's parameter."""
-        for parameter in chain(plan.inputs, plan.outputs, plan.parameters):
-            if parameter.id == self.parameter_id:
-                parameter.actual_value = self.value
-                return
-
-        raise errors.ParameterError(f"Parameter {self.parameter_id} not found on plan {plan.id}.")
+        parameter = plan.get_field_by_id(self.parameter_id)
+        parameter.actual_value = self.value