feat: Milestone 5 - workflow manifest integration and example workflows

databusclient/cli.py

@@ -583,30 +583,67 @@ def workflow():
 
 @workflow.command("run")
 @click.argument("workflow_path", type=click.Path(exists=True, dir_okay=False))
-def workflow_run(workflow_path):
+@click.option(
+    "--manifest",
+    "manifest_path",
+    default=None,
+    help="Write a unified JSON-LD manifest of the entire workflow run to PATH.",
+)
+def workflow_run(workflow_path, manifest_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).
+    step's on_error behavior (fail/continue/retry). Prints a console
+    summary after every run. Use --manifest to also write a unified
+    JSON-LD manifest covering every step.
     """
     try:
         parsed = parse_workflow(workflow_path)
     except WorkflowParseError as e:
         raise click.ClickException(str(e))
 
-    context = StepContext()
-    engine = WorkflowEngine(context=context)
+    # CLI flag takes priority; falls back to the YAML file's own
+    # top-level 'manifest:' key if --manifest was not given on the
+    # command line.
+    if manifest_path is None:
+        manifest_path = parsed.get("manifest")
 
+    # A workflow-level manifest is always built internally (for the
+    # automatic console summary), even when no manifest path is set.
+    # It's only written to disk when a path is provided (via --manifest
+    # or the YAML file's own 'manifest:' key).
+    manifest_ctx = ManifestContext(command="workflow")
+
+    step_context = StepContext()
+    engine = WorkflowEngine(context=step_context, manifest_context=manifest_ctx)
+
+    workflow_error = None
     try:
-        results = engine.run(parsed["steps"])
+        engine.run(parsed["steps"])
     except WorkflowExecutionError as e:
-        raise click.ClickException(str(e))
+        workflow_error = e
+    finally:
+        click.echo("Workflow complete." if workflow_error is None else "Workflow failed.")
+        for result in engine.results:
+            click.echo(f"  {result.name}: {result.status}")
+
+        click.echo("")
+        click.echo(format_summary(ManifestWriter.build_manifest_dict(manifest_ctx)))
+
+        if manifest_path:
+            try:
+                actual_path = ManifestWriter.write(manifest_ctx, manifest_path)
+                click.echo(f"\nManifest written to {actual_path}")
+            except (OSError, IOError) as e:
+                click.echo(
+                    f"WARNING: Manifest could not be written to {manifest_path}: {e}",
+                    err=True,
+                )
 
-    click.echo("Workflow complete.")
-    for result in results:
-        click.echo(f"  {result.name}: {result.status}")
+    if workflow_error is not None:
+        raise click.ClickException(str(workflow_error))
 
 if __name__ == "__main__":
     app()

databusclient/manifest/context.py

@@ -156,4 +156,25 @@ def summary(self) -> dict:
             "succeeded": succeeded,
             "failed": failed,
             "total_bytes": total_bytes,
-        }
+        }
+
+    def merge_from(self, other: "ManifestContext", step_name: Optional[str] = None) -> None:
+        """Merge another context's recorded files into this one.
+
+        Used by the workflow engine: each step records into its own
+        temporary ManifestContext (so per-step failures/successes stay
+        isolated), then that context's entries are merged into the
+        workflow-level master context here, tagged with which step
+        produced them.
+
+        Args:
+            other: The ManifestContext to merge entries from.
+            step_name: If given, tags each merged file entry with
+                "step": step_name, so a multi-step workflow manifest
+                remains traceable to which step produced which file.
+        """
+        for entry in other.files:
+            merged_entry = dict(entry)
+            if step_name is not None:
+                merged_entry["step"] = step_name
+            self.files.append(merged_entry)

databusclient/manifest/writer.py

@@ -30,31 +30,14 @@ class ManifestWriter:
     """Serializes a ManifestContext to a JSON-LD manifest file."""
 
     @staticmethod
-    def write(context: ManifestContext, path: str) -> str:
-        """Write the manifest to a JSON-LD file at the given path.
-
-        Creates parent directories if they do not exist.
-        If a file already exists at `path`, auto-suffixes with _1, _2, etc.
-        and prints a warning rather than silently overwriting.
-        On failure, raises OSError — callers should catch and warn.
-
-        Args:
-            context: The completed ManifestContext to serialize.
-            path: File path to write the manifest to.
-
-        Raises:
-            OSError: If the file cannot be written, or if path is a directory.
+    def build_manifest_dict(context: ManifestContext) -> dict:
+        """Build the JSON-LD manifest dict from a context, without writing
+        to disk. Extracted from write() so callers (like the workflow
+        engine's automatic console summary) can get the dict without
+        needing a file path.
         """
-        if path.endswith(("/", "\\")) or os.path.isdir(path):
-            stripped = path.rstrip("/\\")
-            raise OSError(
-                f"--manifest path '{path}' is a directory, not a file. "
-                f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'."
-            )
-
         summary = context.summary()
 
-        # Build file entries using DataID vocabulary
         file_entries = []
         for f in context.files:
             entry: dict = {
@@ -79,6 +62,8 @@ def write(context: ManifestContext, path: str) -> str:
                 entry["dbus:errorTraceback"] = f["error_traceback"]
             if f.get("retry_count"):
                 entry["dbus:retryCount"] = f["retry_count"]
+            if f.get("step"):
+                entry["dbus:stepName"] = f["step"]
             file_entries.append(entry)
 
         manifest = {
@@ -121,6 +106,33 @@ def write(context: ManifestContext, path: str) -> str:
                 "dbus:errorTraceback": context.operation_error["error_traceback"],
             }
 
+        return manifest
+
+    @staticmethod
+    def write(context: ManifestContext, path: str) -> str:
+        """Write the manifest to a JSON-LD file at the given path.
+
+        Creates parent directories if they do not exist.
+        If a file already exists at `path`, auto-suffixes with _1, _2, etc.
+        and prints a warning rather than silently overwriting.
+        On failure, raises OSError — callers should catch and warn.
+
+        Args:
+            context: The completed ManifestContext to serialize.
+            path: File path to write the manifest to.
+
+        Raises:
+            OSError: If the file cannot be written, or if path is a directory.
+        """
+        if path.endswith(("/", "\\")) or os.path.isdir(path):
+            stripped = path.rstrip("/\\")
+            raise OSError(
+                f"--manifest path '{path}' is a directory, not a file. "
+                f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'."
+            )
+
+        manifest = ManifestWriter.build_manifest_dict(context)
+
         parent = os.path.dirname(os.path.abspath(path))
         if parent:
             os.makedirs(parent, exist_ok=True)

databusclient/workflow/engine.py

@@ -10,8 +10,9 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
+from databusclient.manifest.context import ManifestContext
 from databusclient.workflow.context import StepContext
 from databusclient.workflow.steps import STEP_REGISTRY
 
@@ -32,10 +33,25 @@ def __init__(self, name: str, status: str, error: Exception | None = None,
 
 
 class WorkflowEngine:
-    """Runs a parsed workflow's steps in order, handling errors per step."""
-
-    def __init__(self, context: StepContext | None = None) -> None:
+    """Runs a parsed workflow's steps in order, handling errors per step.
+
+    If manifest_context is given, one unified manifest is built for the
+    entire workflow run: each step gets its own temporary ManifestContext
+    (isolating its recorded files/errors), which is merged into
+    manifest_context afterward, tagged with the step's name. This lets a
+    single workflow manifest remain traceable to which step produced or
+    failed on which file, without touching download.py/deploy.py/delete.py
+    at all -- those already accept manifest_context=None as a no-op, and
+    here they simply receive a real (temporary, per-step) one instead.
+    """
+
+    def __init__(
+        self,
+        context: StepContext | None = None,
+        manifest_context: Optional[ManifestContext] = None,
+    ) -> None:
         self.context = context or StepContext()
+        self.manifest_context = manifest_context
         self.results: List[StepResult] = []
 
     def run(self, steps: List[Dict[str, Any]]) -> List[StepResult]:
@@ -78,32 +94,75 @@ def _run_step_with_error_handling(self, step_config: Dict[str, Any]) -> StepResu
         if on_error == "retry":
             return self._run_with_retry(name, step, step_config)
 
+        step_manifest_ctx = self._start_step_manifest(command)
         try:
             step.run(step_config, self.context)
+            self._finish_step_manifest(step_manifest_ctx, name)
             return StepResult(name, "success")
         except Exception as exc:
+            self._finish_step_manifest(step_manifest_ctx, name, error=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 _start_step_manifest(self, command: str) -> Optional[ManifestContext]:
+        """If a workflow-level manifest is active, give this step its own
+        temporary ManifestContext to record into. Returns None if no
+        workflow manifest was requested -- in that case self.context's
+        manifest_context is left as whatever it already was (e.g. a step's
+        own throwaway context, like DownloadStep uses for output_urls).
+        """
+        if self.manifest_context is None:
+            return None
+        step_ctx = ManifestContext(command=command)
+        self.context.manifest_context = step_ctx
+        return step_ctx
+
+    def _finish_step_manifest(
+        self,
+        step_manifest_ctx: Optional[ManifestContext],
+        step_name: str,
+        error: Optional[Exception] = None,
+    ) -> None:
+        """Merge a completed step's temporary manifest entries into the
+        workflow-level master manifest, tagged with the step name. If the
+        step failed, also record a synthetic entry so the failure is
+        visible in the manifest even if the step recorded no per-file
+        entries before failing.
+        """
+        if self.manifest_context is None or step_manifest_ctx is None:
+            return
+        self.manifest_context.merge_from(step_manifest_ctx, step_name=step_name)
+        if error is not None:
+            self.manifest_context.record_file(
+                url=f"step:{step_name}",
+                status="failed",
+                error_message=str(error),
+            )
+
     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"]
+        command = step_config["command"]
 
         last_error: Exception | None = None
         for attempt in range(1, max_attempts + 1):
+            step_manifest_ctx = self._start_step_manifest(command)
             try:
                 step.run(step_config, self.context)
+                self._finish_step_manifest(step_manifest_ctx, name)
                 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:
+                    self._finish_step_manifest(step_manifest_ctx, name, error=exc)
                 if attempt < max_attempts:
                     time.sleep(delay_seconds)
 

databusclient/workflow/steps.py

@@ -153,6 +153,14 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
         context.set_output(name, "output_files", output_files)
         context.set_output(name, "version_id", resolved["version_id"])
 
+        # deploy()/deploy_from_metadata() do not accept manifest_context
+        # (unlike download()/delete()) -- manifest recording for deploy is
+        # always done manually by the caller. This mirrors exactly what
+        # cli.py's own `deploy` command does after a successful deploy.
+        if context.manifest_context is not None:
+            for url in output_files:
+                context.manifest_context.record_file(url=url, status="success")
+
     def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list:
         files = resolved.get("files")
         if not files:

examples/workflows/README.md

@@ -1,17 +1,28 @@
 # Example Workflows
 
-Three example workflow pipelines, each runnable directly, though the deploy/delete steps use paths under a specific Databus account -- swap in your own account/version paths before running them yourself. All three use real, existing Databus data as their download source.
+Eight example workflow pipelines, each runnable directly (though the deploy/delete steps use paths under a specific Databus account -- swap in your own account/version paths before running them yourself). All use real, existing Databus data as their download source.
 
 ```bash
 export DATABUS_API_KEY=your-key-here
 databusclient workflow run download-deploy.yml
 ```
 
+## Basic examples
+
 - **`download-deploy.yml`** - downloads a real Databus dataset, then redeploys it exactly as downloaded (classic deploy mode, using `${steps.name.output_urls}` - the actual, redirect-resolved source URL, not the local file).
 - **`download-delete.yml`** - downloads a real Databus dataset, then deletes that same version, demonstrating a realistic archive-then-delete workflow.
-- **`full-pipeline.yml`** - chains all three commands together: download a real Databus dataset, deploy it, then delete that same deployed version, demonstrating a complete download-deploy-cleanup pipeline.
+- **`full-pipeline.yml`** - chains all three commands together: download a real Databus dataset, deploy it, then delete that same deployed version.
 
-All three set `api_key: ${DATABUS_API_KEY}` - set that environment variable before running, rather than writing a real key into the file.
+## The five proposal use cases (Milestone 5)
 
-See the main [README's Workflow section](../../README.md#cli-workflow) for the full YAML format, step chaining, error handling, and WebDAV deploy mode documentation.
+- **`reproducible-research-download.yml`** - downloads a dataset with checksum validation and a saved manifest, so the exact same download can be verified or reproduced later.
+- **`nightly-publishing-pipeline.yml`** - download, deploy, then clean up an old version, meant to run unattended (e.g. via cron), with a unified manifest for the whole run.
+- **`batch-deployment-with-retry.yml`** - deploys multiple versions in one run, with `on_error: retry` configured on each deploy step to handle transient failures automatically.
+- **`ci-cd-integration.yml`** - a workflow with no interactive prompts anywhere, safe to call as a step in a CI/CD pipeline such as GitHub Actions.
+- **`failure-debugging.yml`** - intentionally fails a deploy step (invalid API key) to demonstrate what a failed workflow's console output and manifest look like.
+
+## Manifests
 
+Workflows can write a unified manifest covering every step in two ways: pass `--manifest path.jsonld` on the command line, or set a top-level `manifest:` key inside the YAML file itself (the command-line flag takes priority if both are given). Several of the examples above use the YAML key. Every manifest file entry that came from a workflow step is tagged with `dbus:stepName`, so a multi-step run stays traceable to which step produced or failed on which file.
+
+See the main [README's Workflow section](../../README.md#cli-workflow) for the full YAML format, step chaining, error handling, and WebDAV deploy mode documentation.

examples/workflows/batch-deployment-with-retry.yml

@@ -0,0 +1,42 @@
+manifest: ./manifests/batch-deploy.jsonld
+steps:
+  - name: fetch_source
+    command: download
+    uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0
+    localdir: ./workflow-output/batch
+
+  - name: deploy_batch_1
+    command: deploy
+    version_id: https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-1/1.0
+    title: "Batch Deploy Demo 1"
+    abstract: "Batch deployment with retry example"
+    description: "First of several deploys in a batch, demonstrating retry on transient failure"
+    license: https://creativecommons.org/licenses/by-sa/3.0/
+    api_key: ${DATABUS_API_KEY}
+    files: ${steps.fetch_source.output_urls}
+    on_error: retry
+    retry:
+      max_attempts: 3
+      delay_seconds: 5
+
+  - name: deploy_batch_2
+    command: deploy
+    version_id: https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-2/1.0
+    title: "Batch Deploy Demo 2"
+    abstract: "Batch deployment with retry example"
+    description: "Second of several deploys in a batch, demonstrating retry on transient failure"
+    license: https://creativecommons.org/licenses/by-sa/3.0/
+    api_key: ${DATABUS_API_KEY}
+    files: ${steps.fetch_source.output_urls}
+    on_error: retry
+    retry:
+      max_attempts: 3
+      delay_seconds: 5
+
+  - name: cleanup_batch
+    command: delete
+    uris:
+      - https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-1/1.0
+      - https://databus.dbpedia.org/DhanashreeP/test-group/batch-demo-2/1.0
+    api_key: ${DATABUS_API_KEY}
+    on_error: continue