Added provenance_manager extension point

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `provenance_manager` extension point ([#714](https://github.com/alpha-unito/streamflow/pull/714))
 - Enforce `CHANGELOG.md` updates in PRs ([#1066](https://github.com/alpha-unito/streamflow/pull/1066))
 - Add MPI application tutorial and docs agent skills ([#1042](https://github.com/alpha-unito/streamflow/pull/1042))
 - Add fault tolerance paper reference in the documentation ([#1049](https://github.com/alpha-unito/streamflow/pull/1049))

pyproject.toml

@@ -140,6 +140,7 @@ zip-safe = true
 "streamflow.deployment.connector" = ["schemas/base/*.json", "schemas/*.json"]
 "streamflow.deployment.filter" = ["schemas/*.json"]
 "streamflow.persistence" = ["schemas/*.sql", "schemas/*.json"]
+"streamflow.provenance" = ["schemas/**/*.json"]
 "streamflow.recovery" = ["schemas/*.json"]
 "streamflow.scheduling" = ["schemas/*.json"]
 "streamflow.scheduling.policy" = ["schemas/*.json"]

streamflow/ext/plugin.py

@@ -8,6 +8,7 @@
 from streamflow.core.data import DataManager
 from streamflow.core.deployment import BindingFilter, Connector, DeploymentManager
 from streamflow.core.persistence import Database
+from streamflow.core.provenance import ProvenanceManager
 from streamflow.core.recovery import CheckpointManager, FailureManager
 from streamflow.core.scheduling import Policy, Scheduler
 from streamflow.cwl.requirement.docker import cwl_docker_translator_classes
@@ -18,6 +19,7 @@
 from streamflow.deployment.filter import binding_filter_classes
 from streamflow.log_handler import logger
 from streamflow.persistence import database_classes
+from streamflow.provenance import provenance_manager_classes
 from streamflow.recovery import checkpoint_manager_classes, failure_manager_classes
 from streamflow.scheduling import scheduler_classes
 from streamflow.scheduling.policy import policy_classes
@@ -38,6 +40,7 @@
     "deployment_manager": deployment_manager_classes,
     "failure_manager": failure_manager_classes,
     "policy": policy_classes,
+    "provenance_manager": provenance_manager_classes,
     "scheduler": scheduler_classes,
 }
 
@@ -98,6 +101,11 @@ def register_failure_manager(self, name: str, cls: type[FailureManager]) -> None
     def register_policy(self, name: str, cls: type[Policy]) -> None:
         self._register(name, cls, "policy")
 
+    def register_provenance_manager(
+        self, name: str, cls: type[ProvenanceManager]
+    ) -> None:
+        self._register(name, cls, "provenance_manager")
+
     def register_scheduler(self, name: str, cls: type[Scheduler]) -> None:
         self._register(name, cls, "scheduler")
 

streamflow/main.py

@@ -32,7 +32,7 @@
 from streamflow.parser import parser
 from streamflow.persistence import database_classes
 from streamflow.persistence.loading_context import DefaultDatabaseLoadingContext
-from streamflow.provenance import prov_classes
+from streamflow.provenance import provenance_manager_classes
 from streamflow.recovery import checkpoint_manager_classes, failure_manager_classes
 from streamflow.scheduling import scheduler_classes
 from streamflow.version import VERSION
@@ -114,6 +114,7 @@ async def _async_plugin(args: argparse.Namespace) -> None:
 
 
 async def _async_prov(args: argparse.Namespace) -> None:
+    load_extensions()
     context = _get_context_from_config(args.file)
     try:
         db_context = DefaultDatabaseLoadingContext(database=context.database)
@@ -130,27 +131,19 @@ async def _async_prov(args: argparse.Namespace) -> None:
                 )
             )
         )
-        wf_type = {w.type for w in workflows}
-        if len(wf_type) != 1:
+        if len(wf_type := {w.type for w in workflows}) != 1:
             raise WorkflowProvenanceException(
                 "Cannot mix different provenance types in the same file. "
                 f"Workflow {args.workflow} is associated to the following types: {','.join(wf_type)}"
             )
-        wf_type = next(iter(wf_type))
-        if args.type not in prov_classes:
+        if args.type not in provenance_manager_classes:
             raise WorkflowProvenanceException(
                 f"{args.type} provenance format is not supported."
             )
-        elif wf_type not in prov_classes[args.type]:
-            raise WorkflowProvenanceException(
-                "{} provenance format is not supported for workflows of type {}.".format(
-                    args.type, wf_type
-                )
-            )
         else:
-            provenance_manager: ProvenanceManager = prov_classes[args.type][wf_type](
-                context, db_context, workflows
-            )
+            provenance_manager: ProvenanceManager = provenance_manager_classes[
+                args.type
+            ](context, db_context, workflows)
             await provenance_manager.create_archive(
                 outdir=args.outdir,
                 filename=args.name,

streamflow/parser.py

@@ -166,10 +166,9 @@ def __call__(
 prov_parser.add_argument(
     "--type",
     "-t",
-    default="run_crate",
+    default="run_crate/cwl",
     type=str,
-    choices=["run_crate"],
-    help="The type of provenance archive to generate (default: run_crate)",
+    help="The type of provenance archive to generate (default: run_crate/cwl)",
 )
 
 # streamflow report

streamflow/provenance/__init__.py

@@ -5,6 +5,6 @@
 from streamflow.core.provenance import ProvenanceManager
 from streamflow.provenance.run_crate import CWLRunCrateProvenanceManager
 
-prov_classes: MutableMapping[str, MutableMapping[str, type[ProvenanceManager]]] = {
-    "run_crate": {"cwl": CWLRunCrateProvenanceManager}
+provenance_manager_classes: MutableMapping[str, type[ProvenanceManager]] = {
+    "run_crate/cwl": CWLRunCrateProvenanceManager
 }

streamflow/provenance/run_crate.py

@@ -12,6 +12,7 @@
 import uuid
 from abc import ABC, abstractmethod
 from collections.abc import Container, MutableMapping, MutableSequence
+from importlib.resources import files
 from typing import Any, cast
 from zipfile import ZipFile
 
@@ -1471,6 +1472,16 @@ async def get_main_entity(self) -> MutableMapping[str, Any]:
                     )
         return main_entity
 
+    @classmethod
+    def get_schema(cls) -> str:
+        return (
+            files(__package__)
+            .joinpath("schemas")
+            .joinpath("run_crate")
+            .joinpath("cwl.json")
+            .read_text("utf-8")
+        )
+
     async def get_property_value(
         self, name: str, token: Token
     ) -> MutableMapping[str, Any] | None:

streamflow/provenance/schemas/run_crate/cwl.json

@@ -0,0 +1,7 @@
+{
+  "$schema": "https://json-schema.org/draft/2019-09/schema",
+  "$id": "https://streamflow.di.unito.it/schemas/provenance/run_crate/cwl.json",
+  "type": "object",
+  "properties": {},
+  "additionalProperties": false
+}