Merge pull request #153 from fair-workflows/independent-step-publications

Independent step publications

Author: svenvanderburg

fairworkflows/fairstep.py

@@ -4,16 +4,12 @@
 from typing import Callable, get_type_hints, List, Union
 from urllib.parse import urldefrag
 
+import noodles
 import rdflib
 from rdflib import RDF, RDFS, DCTERMS
 
-import noodles
-
 from fairworkflows import namespaces
-from fairworkflows.rdf_wrapper import RdfWrapper
-
-FAIRSTEP_PREDICATES = [RDF.type, namespaces.PPLAN.hasInputVar,
-                       namespaces.PPLAN.hasOutputVar, DCTERMS.description, RDFS.label]
+from fairworkflows.rdf_wrapper import RdfWrapper, replace_in_rdf
 
 
 class FairVariable:
@@ -98,6 +94,7 @@ def __init__(self, label: str = None, description: str = None, uri=None,
         if outputs is not None:
             self.outputs = outputs
         self._is_modified = False
+        self._workflows = set()
 
     @classmethod
     def from_rdf(cls, rdf, uri, fetch_references: bool = False, force: bool = False,
@@ -338,6 +335,17 @@ def validate(self, shacl=False):
         if shacl:
             self.shacl_validate()
 
+    def register_workflow(self, workflow):
+        """Register workflow that this step is part of."""
+        self._workflows.add(workflow)
+
+    def _update_registered_workflows(self):
+        """Update the workflows that this step is part of.
+        NB: it could be that a step was deleted from a workflow
+        """
+        self._workflows = {workflow for workflow in self._workflows
+                           if self in workflow}
+
     def publish_as_nanopub(self, use_test_server=False, **kwargs):
         """
         Publish this rdf as a nanopublication.
@@ -352,7 +360,23 @@ def publish_as_nanopub(self, use_test_server=False, **kwargs):
         Returns:
             a dictionary with publication info, including 'nanopub_uri', and 'concept_uri'
         """
-        return self._publish_as_nanopub(use_test_server=use_test_server, **kwargs)
+        self._update_registered_workflows()
+        old_uri = self.uri
+        self._publish_as_nanopub(use_test_server=use_test_server, **kwargs)
+        var_names = [var.name for var in (self.inputs + self.outputs)]
+        for workflow in self._workflows:
+            replace_in_rdf(workflow.rdf, oldvalue=rdflib.URIRef(old_uri),
+                           newvalue=rdflib.URIRef(self.uri))
+
+            # Similarly replace old URIs for variable name bindings
+            published_step_uri_defrag, _ = urldefrag(self.uri)
+            for var_name in var_names:
+                old_var_uri = old_uri + '#' + var_name
+                new_var_uri = published_step_uri_defrag + '#' + var_name
+                replace_in_rdf(self.rdf, oldvalue=rdflib.URIRef(old_var_uri),
+                               newvalue=rdflib.URIRef(new_var_uri))
+            del workflow._steps[old_uri]
+            workflow._steps[self.uri] = self
 
     def __str__(self):
         """

fairworkflows/fairworkflow.py

@@ -1,21 +1,19 @@
+import inspect
 import warnings
 from copy import deepcopy
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Iterator, Optional
-import inspect
-from urllib.parse import urldefrag
 
 import networkx as nx
 import rdflib
 from rdflib import RDF, RDFS, DCTERMS
 from rdflib.tools.rdf2dot import rdf2dot
 from requests import HTTPError
 
-
 from fairworkflows import namespaces
 from fairworkflows.fairstep import FairStep
-from fairworkflows.rdf_wrapper import RdfWrapper, replace_in_rdf
+from fairworkflows.rdf_wrapper import RdfWrapper
 import nanopub
 
 class FairWorkflow(RdfWrapper):
@@ -54,8 +52,7 @@ def from_rdf(cls, rdf: rdflib.Graph, uri: str,
                 possibly it's associated steps. Should use plex ontology.
             uri: URI of the workflow
             fetch_references: toggles fetching steps. I.e. if we encounter steps
-                that are part of the workflow, but are not specified in the
-                RDF we try fetching them from nanopub
+                that are part of the workflow we try fetching them from nanopub
             force: Toggle forcing creation of object even if url is not in any of the subjects of
                 the passed RDF
             remove_irrelevant_triples: Toggle removing irrelevant triples for this FairWorkflow.
@@ -106,19 +103,18 @@ def from_noodles_promise(cls, noodles_promise, description: str = None, label: s
 
         return self
 
-    def _extract_steps(self, rdf, uri, fetch_steps=False):
+    def _extract_steps(self, rdf, uri, fetch_steps=True):
         """Extract FairStep objects from rdf.
 
-        Create FairStep objects for all steps in the passed RDF. Removes
-        triples describing steps from rdf, those will be represented in
-        the separate step RDF. Optionally try to fetch steps from nanopub.
+        Create FairStep objects for all steps in the passed RDF.
+        Optionally try to fetch steps from nanopub.
         """
         step_refs = rdf.subjects(predicate=namespaces.PPLAN.isStepOfPlan,
                                  object=rdflib.URIRef(uri))
         for step_ref in step_refs:
             step_uri = str(step_ref)
-            step = self._extract_step_from_rdf(step_uri, rdf)
-            if step is None and fetch_steps:
+            step = None
+            if fetch_steps:
                 step = self._fetch_step(uri=step_uri)
             if step is None:
                 warnings.warn(f'Could not get detailed information for '
@@ -153,19 +149,6 @@ def _get_relevant_triples(uri, rdf):
             g.add(triple)
         return g
 
-    @staticmethod
-    def _extract_step_from_rdf(uri, rdf: rdflib.Graph()) -> Optional[FairStep]:
-        relevant_triples = FairStep._get_relevant_triples(uri, rdf)
-        step_rdf = rdflib.Graph()
-        for triple in relevant_triples:
-            step_rdf.add(triple)
-            rdf.remove(triple)
-
-        if len(step_rdf) > 0:
-            return FairStep.from_rdf(step_rdf, uri=uri, remove_irrelevant_triples=False)
-        else:
-            return None
-
     @staticmethod
     def _fetch_step(uri: str) -> Optional[FairStep]:
         try:
@@ -207,6 +190,7 @@ def _add_step(self, step: FairStep):
         self._rdf.add((rdflib.URIRef(step.uri), namespaces.PPLAN.isStepOfPlan,
                        self.self_ref))
         self._last_step_added = step
+        step.register_workflow(self)
 
     def add(self, step: FairStep, follows: FairStep = None):
         """Add a step.
@@ -472,8 +456,10 @@ def draw(self, filepath):
     def publish_as_nanopub(self, use_test_server=False, **kwargs):
         """Publish to nanopub server.
 
-        First publish the steps, use the URIs of the published steps in the workflow. Then
-        publish the workflow.
+        Publish the workflow as nanopublication to the nanopub server.
+
+        Raises:
+            RuntimeError: If one of the steps of the workflow was not published yet.
 
         Args:
             use_test_server (bool): Toggle using the test nanopub server.
@@ -489,25 +475,8 @@ def publish_as_nanopub(self, use_test_server=False, **kwargs):
         for step in self:
             if step.is_modified or not step._is_published:
                 self._is_modified = True  # If one of the steps is modified the workflow is too.
-                old_uri = step.uri
-                var_names = [var.name for var in (step.inputs + step.outputs)]
-
-                step.publish_as_nanopub(use_test_server=use_test_server, **kwargs)
-                published_step_uri = step.uri
-
-                replace_in_rdf(self.rdf, oldvalue=rdflib.URIRef(old_uri),
-                               newvalue=rdflib.URIRef(published_step_uri))
-
-                # Similarly replace old URIs for variable name bindings
-                published_step_uri_defrag, _ = urldefrag(published_step_uri)
-                for var_name in var_names:
-                    old_var_uri = old_uri + '#' + var_name
-                    new_var_uri = published_step_uri_defrag + '#' + var_name
-                    replace_in_rdf(self.rdf, oldvalue=rdflib.URIRef(old_var_uri),
-                                  newvalue=rdflib.URIRef(new_var_uri))
+                raise RuntimeError(f'{step} was not published yet, please publish steps first')
 
-                del self._steps[old_uri]
-                self._steps[published_step_uri] = step
         return self._publish_as_nanopub(use_test_server=use_test_server, **kwargs)
 
     def __str__(self):

fairworkflows/rdf_wrapper.py

@@ -1,5 +1,4 @@
 import warnings
-
 from typing import List
 from urllib.parse import urldefrag
 
@@ -8,8 +7,8 @@
 from nanopub import Publication, NanopubClient
 
 from fairworkflows import namespaces
-
 from fairworkflows.config import PACKAGE_DIR
+
 PLEX_SHAPES_SHACL_FILEPATH = str(PACKAGE_DIR / 'resources' / 'plex-shapes.ttl')
 
 

…urces/test_workflow_excluding_steps.trig tests/resources/test_workflow.trigtests/resources/test_workflow_excluding_steps.trig renamed to tests/resources/test_workflow.trig tests/resources/test_workflow_excluding_steps.trig renamed to tests/resources/test_workflow.trig

@@ -1,13 +1,14 @@
 @prefix this: <http://www.example.org/workflow1> .
-@prefix sub: <http://www.example.org/workflow1#> .
 @prefix pplan: <http://purl.org/net/p-plan#> .
 @prefix terms: <http://purl.org/dc/terms/> .
 @prefix pwo: <http://purl.org/spar/pwo/> .
+@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
 
 {
     this: a pplan:Plan ;
        terms:description "This is a test workflow." ;
-       pwo:hasFirstStep sub:step1 .
+       pwo:hasFirstStep <http://www.example.org/step1> ;
+       rdfs:label 'Test workflow' .
 
-    sub:step1 pplan:isStepOfPlan this: .
+    <http://www.example.org/step1> pplan:isStepOfPlan this: .
 }