gsoc26: Declarative Workflow Engine (chaining download, deploy, delete)

.gitignore

@@ -169,4 +169,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
-
+workflow-output/

databusclient/cli.py

@@ -13,6 +13,9 @@
 from databusclient.manifest.replay import ManifestReplayError, replay_manifest, load_manifest
 from databusclient.manifest.summary import format_summary
 from databusclient.extensions import webdav
+from databusclient.workflow.parser import WorkflowParseError, parse_workflow
+from databusclient.workflow.engine import WorkflowEngine, WorkflowExecutionError
+from databusclient.workflow.context import StepContext
 
 
 @click.group()
@@ -568,5 +571,42 @@ def manifest_summary(manifest_path):
     except ManifestReplayError as e:
         raise click.ClickException(str(e))
 
+@app.group()
+def workflow():
+    """
+    Workflow utilities.
+
+    Run multi-step download/deploy/delete pipelines defined in YAML.
+    """
+    pass
+
+
+@workflow.command("run")
+@click.argument("workflow_path", type=click.Path(exists=True, dir_okay=False))
+def workflow_run(workflow_path):
+    """
+    Run a declarative workflow pipeline from a YAML file.
+
+    Executes each step in order, chaining outputs between steps via
+    ${steps.name.output_files}-style references, and applying each
+    step's on_error behavior (fail/continue/retry).
+    """
+    try:
+        parsed = parse_workflow(workflow_path)
+    except WorkflowParseError as e:
+        raise click.ClickException(str(e))
+
+    context = StepContext()
+    engine = WorkflowEngine(context=context)
+
+    try:
+        results = engine.run(parsed["steps"])
+    except WorkflowExecutionError as e:
+        raise click.ClickException(str(e))
+
+    click.echo("Workflow complete.")
+    for result in results:
+        click.echo(f"  {result.name}: {result.status}")
+
 if __name__ == "__main__":
     app()

databusclient/workflow/__init__.py

@@ -0,0 +1,4 @@
+"""Workflow engine for the Databus Python Client.
+
+Orchestrates multi-step download/deploy/delete pipelines defined in YAML.
+"""

databusclient/workflow/context.py

@@ -0,0 +1,100 @@
+"""StepContext — tracks step outputs and resolves ${steps.name.key} references at runtime.
+
+WorkflowParser resolves ${VAR_NAME} environment variables at parse time,
+but deliberately leaves ${steps.step_name.output_files} tokens untouched,
+since those values don't exist until the referenced step has actually run.
+StepContext is what resolves them, once the WorkflowEngine has executed
+each step in order.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict
+
+# Matches a single ${steps.step_name.key} token.
+_STEP_REF_RE = re.compile(r"\$\{steps\.([^.}]+)\.([^}]+)\}")
+
+
+class StepReferenceError(Exception):
+    """Raised when a ${steps.name.key} reference cannot be resolved."""
+
+
+class StepContext:
+    """Stores per-step outputs and resolves ${steps.name.key} references.
+
+    manifest_context is accepted but unused in Milestone 4 -- it exists as
+    a seam so Milestone 5 can wire in manifest recording without changing
+    this class's structure. When None, it has zero effect, matching the
+    manifest_context=None pattern already used throughout download.py,
+    deploy.py, and delete.py.
+    """
+
+    def __init__(self, manifest_context=None) -> None:
+        self._outputs: Dict[str, Dict[str, Any]] = {}
+        self.manifest_context = manifest_context
+
+    def set_output(self, step_name: str, key: str, value: Any) -> None:
+        """Record an output value produced by a step.
+
+        Args:
+            step_name: Name of the step that produced this output.
+            key: Output key, e.g. "output_files".
+            value: The value to store (e.g. a list of file paths).
+        """
+        self._outputs.setdefault(step_name, {})[key] = value
+
+    def get_output(self, step_name: str, key: str) -> Any:
+        """Retrieve a previously recorded output value.
+
+        Raises:
+            StepReferenceError: If the step or key is unknown.
+        """
+        if step_name not in self._outputs:
+            raise StepReferenceError(
+                f"Reference to unknown or not-yet-executed step '{step_name}'."
+            )
+        if key not in self._outputs[step_name]:
+            raise StepReferenceError(
+                f"Step '{step_name}' has no recorded output '{key}'. "
+                f"Available outputs: {sorted(self._outputs[step_name].keys())}."
+            )
+        return self._outputs[step_name][key]
+
+    def resolve(self, value: Any) -> Any:
+        """Recursively resolve ${steps.name.key} references in a value.
+
+        A value that is EXACTLY a single ${steps.name.key} token (nothing
+        else in the string) resolves to the raw stored value (e.g. a list),
+        preserving its type. A token embedded inside a larger string is
+        resolved by inserting str(value) in place, same as environment
+        variable substitution.
+
+        Args:
+            value: A string, list, dict, or scalar value from a step config.
+
+        Returns:
+            The value with all ${steps.*} references resolved.
+
+        Raises:
+            StepReferenceError: If a referenced step/key is unknown.
+        """
+        if isinstance(value, str):
+            full_match = _STEP_REF_RE.fullmatch(value)
+            if full_match:
+                step_name, key = full_match.group(1), full_match.group(2)
+                return self.get_output(step_name, key)
+
+            def _replace(match: re.Match) -> str:
+                step_name, key = match.group(1), match.group(2)
+                return str(self.get_output(step_name, key))
+
+            return _STEP_REF_RE.sub(_replace, value)
+
+        if isinstance(value, list):
+            return [self.resolve(item) for item in value]
+
+        if isinstance(value, dict):
+            return {k: self.resolve(v) for k, v in value.items()}
+
+        return value

databusclient/workflow/engine.py

@@ -0,0 +1,110 @@
+"""WorkflowEngine — executes a sequence of parsed workflow steps in order.
+
+Applies each step's on_error behavior (fail/continue/retry) around a call
+to the step's run() method. Retries operate at the whole-step level --
+the engine has no visibility into partial failures inside a step (e.g.
+one file out of several failing during a download), since download(),
+deploy(), and delete() are called as single atomic operations.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, List
+
+from databusclient.workflow.context import StepContext
+from databusclient.workflow.steps import STEP_REGISTRY
+
+
+class WorkflowExecutionError(Exception):
+    """Raised when a workflow step fails and on_error is 'fail' (or defaults to it)."""
+
+
+class StepResult:
+    """Outcome of running a single step."""
+
+    def __init__(self, name: str, status: str, error: Exception | None = None,
+                 attempts: int = 1) -> None:
+        self.name = name
+        self.status = status  # "success", "failed", "skipped_error"
+        self.error = error
+        self.attempts = attempts
+
+
+class WorkflowEngine:
+    """Runs a parsed workflow's steps in order, handling errors per step."""
+
+    def __init__(self, context: StepContext | None = None) -> None:
+        self.context = context or StepContext()
+        self.results: List[StepResult] = []
+
+    def run(self, steps: List[Dict[str, Any]]) -> List[StepResult]:
+        """Execute all steps in order.
+
+        Args:
+            steps: List of validated, environment-substituted step dicts
+                (as produced by WorkflowParser.parse_workflow).
+
+        Returns:
+            List of StepResult, one per step actually attempted.
+
+        Raises:
+            WorkflowExecutionError: If a step with on_error 'fail' (the
+                default) ultimately fails.
+        """
+        for step_config in steps:
+            result = self._run_step_with_error_handling(step_config)
+            self.results.append(result)
+            if result.status == "failed":
+                # on_error was 'fail' (or defaulted to it) -- stop the workflow.
+                raise WorkflowExecutionError(
+                    f"Step '{result.name}' failed: {result.error}"
+                )
+        return self.results
+
+    def _run_step_with_error_handling(self, step_config: Dict[str, Any]) -> StepResult:
+        name = step_config["name"]
+        command = step_config["command"]
+        on_error = step_config.get("on_error", "fail")
+
+        step_class = STEP_REGISTRY.get(command)
+        if step_class is None:
+            # Should already be caught by the parser, but defend anyway.
+            raise WorkflowExecutionError(
+                f"Step '{name}' has unknown command '{command}'."
+            )
+        step = step_class()
+
+        if on_error == "retry":
+            return self._run_with_retry(name, step, step_config)
+
+        try:
+            step.run(step_config, self.context)
+            return StepResult(name, "success")
+        except Exception as exc:
+            if on_error == "continue":
+                print(f"WARNING: step '{name}' failed and on_error is 'continue': {exc}")
+                return StepResult(name, "skipped_error", error=exc)
+            # on_error == "fail" (or missing/defaulted to fail)
+            return StepResult(name, "failed", error=exc)
+
+    def _run_with_retry(self, name: str, step: Any, step_config: Dict[str, Any]) -> StepResult:
+        retry_config = step_config["retry"]
+        max_attempts = retry_config["max_attempts"]
+        delay_seconds = retry_config["delay_seconds"]
+
+        last_error: Exception | None = None
+        for attempt in range(1, max_attempts + 1):
+            try:
+                step.run(step_config, self.context)
+                return StepResult(name, "success", attempts=attempt)
+            except Exception as exc:
+                last_error = exc
+                print(
+                    f"WARNING: step '{name}' attempt {attempt}/{max_attempts} "
+                    f"failed: {exc}"
+                )
+                if attempt < max_attempts:
+                    time.sleep(delay_seconds)
+
+        return StepResult(name, "failed", error=last_error, attempts=max_attempts)