Check if bindings exist in the workflow (#708)

This commit adds a check to warn the user if there are bindings
(steps or ports) in the StreamFlow config file that do not have a
corresponding entity in the workflow. This commit also refactors the
`Command` classes in the `streamflow.core` to remove circular
dependencies. Finally, this commit improves the check for circular
deployment wrappers introduced in #705 by keeping track of the
already visited deployments.

Author: LanderOtto

streamflow/config/config.py

@@ -12,6 +12,8 @@
 
 def set_targets(current_node, target):
     for node in current_node["children"].values():
+        if "port" in node:
+            continue
         if "step" not in node:
             node["step"] = target
         set_targets(node, node["step"])
@@ -69,6 +71,8 @@ def _check_stacked_deployments(self) -> None:
                         f"The deployment `{deployment['name']}` leads to a circular reference: "
                         f"Recursive deployment definitions are not allowed."
                     )
+                else:
+                    deployments.add(deployment["name"])
 
     def _process_binding(self, binding: MutableMapping[str, Any]):
         targets = (
@@ -88,6 +92,11 @@ def _process_binding(self, binding: MutableMapping[str, Any]):
             else:
                 raise WorkflowDefinitionException(f"Binding filter {f} is not defined")
         path = PurePosixPath(binding["step"] if "step" in binding else binding["port"])
+        if not path.is_absolute():
+            raise WorkflowDefinitionException(
+                f"Binding {path.as_posix()} is not well-defined in the StreamFlow file. "
+                f"It must be an absolute POSIX path"
+            )
         self.put(path, target_type, config)
 
     def get(

streamflow/core/recovery.py

@@ -10,10 +10,9 @@
 from streamflow.workflow.token import JobToken
 
 if TYPE_CHECKING:
-    from streamflow.core.command import CommandOutput
     from streamflow.core.context import StreamFlowContext
     from streamflow.core.data import DataLocation
-    from streamflow.core.workflow import Job, Step, Token, Workflow
+    from streamflow.core.workflow import CommandOutput, Job, Step, Token, Workflow
 
 
 class CheckpointManager(SchemaEntity):

streamflow/core/workflow.py

@@ -10,19 +10,207 @@
 from typing import TYPE_CHECKING, TypeVar, cast
 
 from streamflow.core import utils
+from streamflow.core.deployment import Connector, ExecutionLocation, Target
 from streamflow.core.exception import WorkflowExecutionException
 from streamflow.core.persistence import (
     DatabaseLoadingContext,
     DependencyType,
     PersistableEntity,
 )
+from streamflow.core.utils import get_class_from_name, get_class_fullname
 
 if TYPE_CHECKING:
     from typing import Any
 
     from streamflow.core.context import StreamFlowContext
 
 
+class Command(ABC):
+    def __init__(self, step: Step):
+        super().__init__()
+        self.step: Step = step
+
+    @abstractmethod
+    async def execute(self, job: Job) -> CommandOutput: ...
+
+    @classmethod
+    async def load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+        step: Step,
+    ) -> Command:
+        type_ = cast(type[Command], utils.get_class_from_name(row["type"]))
+        return await type_._load(context, row["params"], loading_context, step)
+
+    async def save(self, context: StreamFlowContext):
+        return {
+            "type": utils.get_class_fullname(type(self)),
+            "params": await self._save_additional_params(context),
+        }
+
+    @classmethod
+    async def _load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+        step: Step,
+    ):
+        return cls(step=step)
+
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> MutableMapping[str, Any]:
+        return {}
+
+
+class CommandOptions(ABC):
+    pass
+
+
+class CommandOutput:
+    __slots__ = ("value", "status")
+
+    def __init__(self, value: Any, status: Status):
+        self.value: Any = value
+        self.status: Status = status
+
+    def update(self, value: Any):
+        return CommandOutput(value=value, status=self.status)
+
+
+class CommandOutputProcessor(ABC):
+    def __init__(self, name: str, workflow: Workflow, target: Target | None = None):
+        self.name: str = name
+        self.workflow: Workflow = workflow
+        self.target: Target | None = target
+
+    def _get_connector(self, connector: Connector | None, job: Job) -> Connector:
+        return connector or self.workflow.context.scheduler.get_connector(job.name)
+
+    async def _get_locations(
+        self, connector: Connector | None, job: Job
+    ) -> MutableSequence[ExecutionLocation]:
+        if self.target:
+            available_locations = await connector.get_available_locations(
+                service=self.target.service
+            )
+            return [loc.location for loc in available_locations.values()]
+        else:
+            return self.workflow.context.scheduler.get_locations(job.name)
+
+    @classmethod
+    async def _load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+    ) -> CommandOutputProcessor:
+        return cls(
+            name=row["name"],
+            workflow=await loading_context.load_workflow(context, row["workflow"]),
+            target=(
+                (await loading_context.load_target(context, row["workflow"]))
+                if row["target"]
+                else None
+            ),
+        )
+
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> MutableMapping[str, Any]:
+        if self.target:
+            await self.target.save(context)
+        return {
+            "name": self.name,
+            "workflow": self.workflow.persistent_id,
+            "target": self.target.persistent_id if self.target else None,
+        }
+
+    @classmethod
+    async def load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+    ) -> CommandOutputProcessor:
+        type_ = cast(
+            type[CommandOutputProcessor], utils.get_class_from_name(row["type"])
+        )
+        return await type_._load(context, row["params"], loading_context)
+
+    @abstractmethod
+    async def process(
+        self,
+        job: Job,
+        command_output: CommandOutput,
+        connector: Connector | None = None,
+    ) -> Token | None: ...
+
+    async def save(self, context: StreamFlowContext):
+        return {
+            "type": utils.get_class_fullname(type(self)),
+            "params": await self._save_additional_params(context),
+        }
+
+
+class CommandToken:
+    __slots__ = ("name", "position", "value")
+
+    def __init__(self, name: str | None, position: int | None, value: Any):
+        self.name: str | None = name
+        self.position: int | None = position
+        self.value: Any = value
+
+
+class CommandTokenProcessor(ABC):
+    def __init__(self, name: str):
+        self.name: str = name
+
+    @classmethod
+    async def _load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+    ):
+        return cls(name=row["name"])
+
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> MutableMapping[str, Any]:
+        return {"name": self.name}
+
+    @abstractmethod
+    def bind(
+        self,
+        token: Token | None,
+        position: int | None,
+        options: CommandOptions,
+    ) -> CommandToken: ...
+
+    @abstractmethod
+    def check_type(self, token: Token) -> bool: ...
+
+    @classmethod
+    async def load(
+        cls,
+        context: StreamFlowContext,
+        row: MutableMapping[str, Any],
+        loading_context: DatabaseLoadingContext,
+    ) -> CommandTokenProcessor:
+        type_ = cast(type[CommandTokenProcessor], get_class_from_name(row["type"]))
+        return await type_._load(context, row["params"], loading_context)
+
+    async def save(self, context: StreamFlowContext):
+        return {
+            "type": get_class_fullname(type(self)),
+            "params": await self._save_additional_params(context),
+        }
+
+
 class Executor(ABC):
     def __init__(self, workflow: Workflow):
         self.workflow: Workflow = workflow
@@ -514,6 +702,7 @@ def __init__(
         self.context: StreamFlowContext = context
         self.config: MutableMapping[str, Any] = config
         self.name: str = name if name is not None else str(uuid.uuid4())
+        self.input_ports: MutableMapping[str, str] = {}
         self.output_ports: MutableMapping[str, str] = {}
         self.ports: MutableMapping[str, Port] = {}
         self.steps: MutableMapping[str, Step] = {}
@@ -548,6 +737,12 @@ def create_step(self, cls: type[S], name: str = None, **kwargs) -> S:
         self.steps[name] = step
         return step
 
+    def get_input_port(self, name: str) -> Port:
+        return self.ports[self.input_ports[name]]
+
+    def get_input_ports(self) -> MutableMapping[str, Port]:
+        return {name: self.ports[p] for name, p in self.input_ports.items()}
+
     def get_output_port(self, name: str) -> Port:
         return self.ports[self.output_ports[name]]
 

streamflow/cwl/command.py

@@ -16,17 +16,6 @@
 from ruamel.yaml import RoundTripRepresenter
 from ruamel.yaml.scalarfloat import ScalarFloat
 
-from streamflow.core.command import (
-    Command,
-    CommandOptions,
-    CommandToken,
-    CommandTokenProcessor,
-    ListCommandToken,
-    MapCommandTokenProcessor,
-    ObjectCommandToken,
-    ObjectCommandTokenProcessor,
-    TokenizedCommand,
-)
 from streamflow.core.context import StreamFlowContext
 from streamflow.core.data import DataLocation
 from streamflow.core.deployment import Connector
@@ -36,7 +25,17 @@
 )
 from streamflow.core.persistence import DatabaseLoadingContext
 from streamflow.core.utils import flatten_list, get_tag
-from streamflow.core.workflow import Job, Status, Step, Token, Workflow
+from streamflow.core.workflow import (
+    Command,
+    CommandOptions,
+    CommandToken,
+    CommandTokenProcessor,
+    Job,
+    Status,
+    Step,
+    Token,
+    Workflow,
+)
 from streamflow.cwl import utils
 from streamflow.cwl.processor import (
     CWLCommandOutput,
@@ -50,6 +49,13 @@
 from streamflow.data.remotepath import StreamFlowPath
 from streamflow.deployment.utils import get_path_processor
 from streamflow.log_handler import logger
+from streamflow.workflow.command import (
+    ListCommandToken,
+    MapCommandTokenProcessor,
+    ObjectCommandToken,
+    ObjectCommandTokenProcessor,
+    TokenizedCommand,
+)
 from streamflow.workflow.step import ExecuteStep
 from streamflow.workflow.utils import get_token_value
 

streamflow/cwl/processor.py

@@ -9,7 +9,6 @@
 import cwl_utils.file_formats
 from schema_salad.exceptions import ValidationException
 
-from streamflow.core.command import CommandOutput, CommandOutputProcessor
 from streamflow.core.context import StreamFlowContext
 from streamflow.core.deployment import Connector, LocalTarget, Target
 from streamflow.core.exception import (
@@ -18,7 +17,14 @@
 )
 from streamflow.core.persistence import DatabaseLoadingContext
 from streamflow.core.utils import flatten_list, get_tag
-from streamflow.core.workflow import Job, Status, Token, TokenProcessor
+from streamflow.core.workflow import (
+    CommandOutput,
+    CommandOutputProcessor,
+    Job,
+    Status,
+    Token,
+    TokenProcessor,
+)
 from streamflow.cwl import utils
 from streamflow.cwl.token import CWLFileToken
 from streamflow.cwl.utils import LoadListing, SecondaryFile

streamflow/cwl/step.py

@@ -8,7 +8,6 @@
 from collections.abc import MutableMapping, MutableSequence
 from typing import Any, cast
 
-from streamflow.core.command import Command, CommandOutput, CommandOutputProcessor
 from streamflow.core.context import StreamFlowContext
 from streamflow.core.data import DataLocation, DataType
 from streamflow.core.deployment import Connector, ExecutionLocation
@@ -18,7 +17,14 @@
 )
 from streamflow.core.persistence import DatabaseLoadingContext
 from streamflow.core.utils import get_entity_ids, get_tag, random_name
-from streamflow.core.workflow import Job, Port, Token
+from streamflow.core.workflow import (
+    Command,
+    CommandOutput,
+    CommandOutputProcessor,
+    Job,
+    Port,
+    Token,
+)
 from streamflow.cwl import utils
 from streamflow.cwl.token import CWLFileToken
 from streamflow.cwl.utils import (