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/

README.md

@@ -21,6 +21,7 @@ Command-line and Python client for downloading and deploying datasets on DBpedia
   - [Manifest](#cli-manifest)
     - [Replay](#cli-manifest-replay)
     - [Summary](#cli-manifest-summary)
+  - [Workflow](#cli-workflow)
 - [Module Usage](#module-usage)
   - [Deploy](#module-deploy)
 - [Development & Contributing](#development--contributing)
@@ -709,6 +710,124 @@ Status   : completed
 
 Only existing data already stored in the manifest is read — no new files are downloaded or written, and no network access happens.
 
+<a id="cli-workflow"></a>
+### Workflow
+
+The workflow command runs a multi-step pipeline of `download`, `deploy`, and `delete` operations defined in a YAML file. Steps run in order, and a later step can use the output of an earlier step — for example, deploying the exact file a previous step just downloaded.
+
+```bash
+# Python
+databusclient workflow run [OPTIONS] WORKFLOW_PATH
+# Docker
+docker run --rm -v $(pwd):/data dbpedia/databus-python-client workflow run [OPTIONS] WORKFLOW_PATH
+```
+
+**Help and further information on the workflow command:**
+```bash
+# Python
+databusclient workflow run --help
+# Docker
+docker run --rm -v $(pwd):/data dbpedia/databus-python-client workflow run --help
+
+# Output:
+Usage: databusclient workflow run [OPTIONS] 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).
+
+Options:
+  --help  Show this message and exit.
+```
+
+#### Workflow YAML format
+
+A workflow file has a top-level `steps:` list. Each step needs a unique `name` and a `command` (`download`, `deploy`, or `delete`), plus fields specific to that command.
+
+```yaml
+steps:
+  - name: fetch_dataset
+    command: download
+    uri: https://databus.dbpedia.org/dbpedia/mappings/mappingbased-literals/2022.12.01/mappingbased-literals_lang=az.ttl.bz2
+    localdir: ./data
+
+  - name: publish_dataset
+    command: deploy
+    version_id: https://databus.dbpedia.org/myaccount/research/labels/2024.01
+    title: "Processed Labels"
+    abstract: "Processed from DBpedia 2023.12.01"
+    description: "Converted and redeployed labels dataset"
+    license: https://creativecommons.org/licenses/by-sa/3.0/
+    api_key: ${DATABUS_API_KEY}
+    files: ${steps.fetch_dataset.output_urls}
+    on_error: fail
+```
+
+**Environment variables:** any value written as `${VARIABLE_NAME}` is resolved from the environment when the workflow starts. If the variable is not set, the workflow fails immediately with a clear error before any step runs — credentials should always be passed this way, never written directly in the file.
+
+**Step chaining:** a step's outputs can be referenced by later steps using `${steps.step_name.output_key}`:
+- `${steps.name.output_files}` — local file paths produced by a `download` step.
+- `${steps.name.output_urls}` — the actual, redirect-resolved source URL(s) the file was downloaded from, useful for redeploying an unmodified file via classic deploy mode.
+
+#### Deploy step modes within a workflow
+
+A `deploy` step supports the same modes as the `deploy` CLI command:
+
+- **Classic mode** (`files:` is a list of URLs) — use `${steps.name.output_urls}` to redeploy a file exactly as it was downloaded, unmodified. Classic mode does not accept local file paths; if `files:` contains anything other than a `http://`/`https://` URL, the step fails with a clear error rather than crashing.
+- **WebDAV mode** (`webdav_url:`, `remote:`, `path:` all provided) — use `${steps.name.output_files}` (local paths) here. The step uploads the local files to the WebDAV server first, then deploys the resulting URLs. This is the only way to deploy a file that was locally modified during the workflow (e.g. via `--format`/`--compression` on the download step), since only WebDAV mode re-establishes a real, fetchable URL for locally changed content.
+
+```yaml
+  - name: publish_converted_dataset
+    command: deploy
+    version_id: https://databus.dbpedia.org/myaccount/research/labels/2024.01
+    title: "Processed Labels"
+    abstract: "Processed from DBpedia 2023.12.01"
+    description: "Converted and redeployed labels dataset"
+    license: https://creativecommons.org/licenses/by-sa/3.0/
+    api_key: ${DATABUS_API_KEY}
+    webdav_url: https://cloud.example.com/remote.php/webdav
+    remote: nextcloud
+    path: datasets/mydataset
+    files: ${steps.fetch_dataset.output_files}
+```
+
+#### Error handling
+
+Each step declares an `on_error` behavior (defaults to `fail` if not set):
+
+| Mode | Behavior |
+|---|---|
+| `fail` | Stop the entire workflow immediately if this step fails. |
+| `continue` | Log the failure and move on to the next step anyway. |
+| `retry` | Retry the step up to `max_attempts` times, waiting `delay_seconds` between attempts. If all attempts fail, the workflow stops. |
+
+```yaml
+  - name: fetch_dataset
+    command: download
+    uri: https://databus.dbpedia.org/...
+    on_error: retry
+    retry:
+      max_attempts: 3
+      delay_seconds: 5
+```
+
+A retry re-runs the entire step from scratch, not just the part that failed.
+
+**Delete steps never prompt for confirmation inside a workflow** — since workflows are meant to run unattended, a `delete` step always behaves as if `--force` was passed.
+
+#### Examples
+
+Full working example files are available under [`examples/workflows/`](examples/workflows/):
+- `download-deploy.yml` — download a file, then redeploy it (classic mode).
+- `download-delete.yml` — download a file, then delete an old version.
+- `full-pipeline.yml` — download, deploy, and delete chained together in one run.
+
+```bash
+databusclient workflow run examples/workflows/download-deploy.yml
+```
+
 ## Module Usage
 
 <a id="module-deploy"></a>

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