diff --git a/.gitignore b/.gitignore
index 1e3c183..0401074 100644
--- a/.gitignore
+++ b/.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/
\ No newline at end of file
diff --git a/README.md b/README.md
index 148fcb7..4b69453 100644
--- a/README.md
+++ b/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.
+
+### 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
diff --git a/databusclient/cli.py b/databusclient/cli.py
index fab57a9..92989c4 100644
--- a/databusclient/cli.py
+++ b/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()
diff --git a/databusclient/workflow/__init__.py b/databusclient/workflow/__init__.py
new file mode 100644
index 0000000..f560e5f
--- /dev/null
+++ b/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.
+"""
\ No newline at end of file
diff --git a/databusclient/workflow/context.py b/databusclient/workflow/context.py
new file mode 100644
index 0000000..4c0cd6e
--- /dev/null
+++ b/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
\ No newline at end of file
diff --git a/databusclient/workflow/engine.py b/databusclient/workflow/engine.py
new file mode 100644
index 0000000..d5cec54
--- /dev/null
+++ b/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)
\ No newline at end of file
diff --git a/databusclient/workflow/parser.py b/databusclient/workflow/parser.py
new file mode 100644
index 0000000..267dda1
--- /dev/null
+++ b/databusclient/workflow/parser.py
@@ -0,0 +1,188 @@
+"""WorkflowParser — loads and validates a YAML workflow pipeline file.
+
+Parses a YAML file describing a sequence of steps (download/deploy/delete),
+validates its structure, and substitutes environment variables of the form
+${VAR_NAME}. References of the form ${steps.step_name.output_files} are
+left untouched here -- those are resolved at runtime by StepContext once
+each step has actually run, since their values don't exist yet at parse time.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any, Dict, List
+
+import yaml
+
+VALID_COMMANDS = {"download", "deploy", "delete"}
+VALID_ON_ERROR = {"fail", "continue", "retry"}
+
+# Matches ${...} tokens. The captured group is everything between the braces.
+_TOKEN_RE = re.compile(r"\$\{([^}]+)\}")
+
+
+class WorkflowParseError(Exception):
+ """Raised when a workflow YAML file is invalid or fails validation."""
+
+
+class MissingEnvVarError(WorkflowParseError):
+ """Raised when a workflow references an environment variable that is not set."""
+
+
+def _load_yaml(path: str) -> Any:
+ """Load a YAML file using safe_load (never load arbitrary Python objects)."""
+ try:
+ with open(path, "r", encoding="utf-8-sig") as f:
+ return yaml.safe_load(f)
+ except FileNotFoundError as e:
+ raise WorkflowParseError(f"Workflow file not found: {path}") from e
+ except yaml.YAMLError as e:
+ raise WorkflowParseError(f"Workflow file is not valid YAML: {path}\n{e}") from e
+
+
+def _substitute_value(value: Any, step_name: str) -> Any:
+ """Recursively substitute ${VAR_NAME} environment variables in a value.
+
+ Tokens of the form ${steps.*} are left untouched -- they are resolved
+ later, at runtime, by StepContext once earlier steps have produced
+ their outputs. Only non-"steps."-prefixed tokens are treated as
+ environment variables here.
+
+ Args:
+ value: A string, list, dict, or scalar value from the parsed YAML.
+ step_name: Name of the step this value belongs to (for error messages).
+
+ Returns:
+ The value with environment variables substituted.
+
+ Raises:
+ MissingEnvVarError: If a referenced environment variable is not set.
+ """
+ if isinstance(value, str):
+ def _replace(match: re.Match) -> str:
+ token = match.group(1)
+ if token.startswith("steps."):
+ # Leave step-output references untouched for runtime resolution.
+ return match.group(0)
+ env_value = os.environ.get(token)
+ if env_value is None:
+ raise MissingEnvVarError(
+ f"Step '{step_name}' references environment variable "
+ f"'{token}' which is not set."
+ )
+ return env_value
+
+ return _TOKEN_RE.sub(_replace, value)
+
+ if isinstance(value, list):
+ return [_substitute_value(item, step_name) for item in value]
+
+ if isinstance(value, dict):
+ return {k: _substitute_value(v, step_name) for k, v in value.items()}
+
+ return value
+
+
+def _validate_step(step: Any, index: int, seen_names: set) -> Dict[str, Any]:
+ """Validate the generic structure of a single step.
+
+ Only validates fields common to all step types (name, command, on_error,
+ retry config). Command-specific required fields (e.g. 'uri' for download)
+ are validated later, when the step actually executes.
+
+ Args:
+ step: The raw step dict from the parsed YAML.
+ index: Position of this step in the steps list (for error messages).
+ seen_names: Set of step names already seen, for duplicate detection.
+
+ Returns:
+ The validated step dict (unchanged, just checked).
+
+ Raises:
+ WorkflowParseError: If the step is structurally invalid.
+ """
+ if not isinstance(step, dict):
+ raise WorkflowParseError(f"Step at index {index} must be a mapping/object.")
+
+ name = step.get("name")
+ if not name or not isinstance(name, str):
+ raise WorkflowParseError(f"Step at index {index} is missing a valid 'name'.")
+
+ if name in seen_names:
+ raise WorkflowParseError(f"Duplicate step name '{name}'. Step names must be unique.")
+ seen_names.add(name)
+
+ command = step.get("command")
+ if command not in VALID_COMMANDS:
+ raise WorkflowParseError(
+ f"Step '{name}' has invalid command '{command}'. "
+ f"Must be one of: {sorted(VALID_COMMANDS)}."
+ )
+
+ on_error = step.get("on_error", "fail")
+ if on_error not in VALID_ON_ERROR:
+ raise WorkflowParseError(
+ f"Step '{name}' has invalid on_error '{on_error}'. "
+ f"Must be one of: {sorted(VALID_ON_ERROR)}."
+ )
+
+ if on_error == "retry":
+ retry_config = step.get("retry")
+ if not isinstance(retry_config, dict):
+ raise WorkflowParseError(
+ f"Step '{name}' has on_error: retry but is missing a 'retry' "
+ f"configuration block with 'max_attempts' and 'delay_seconds'."
+ )
+ max_attempts = retry_config.get("max_attempts")
+ if not isinstance(max_attempts, int) or max_attempts < 1:
+ raise WorkflowParseError(
+ f"Step '{name}' retry.max_attempts must be a positive integer."
+ )
+ delay_seconds = retry_config.get("delay_seconds")
+ if not isinstance(delay_seconds, (int, float)) or delay_seconds < 0:
+ raise WorkflowParseError(
+ f"Step '{name}' retry.delay_seconds must be a non-negative number."
+ )
+
+ return step
+
+
+def parse_workflow(path: str) -> Dict[str, Any]:
+ """Load, validate, and substitute environment variables in a workflow YAML file.
+
+ Args:
+ path: Path to the workflow YAML file.
+
+ Returns:
+ A dict with keys:
+ "manifest": Optional manifest output path (str or None).
+ "steps": List of validated, environment-substituted step dicts.
+
+ Raises:
+ WorkflowParseError: If the file is missing, invalid YAML, or fails
+ structural validation.
+ MissingEnvVarError: If a step references an unset environment variable.
+ """
+ raw = _load_yaml(path)
+
+ if not isinstance(raw, dict):
+ raise WorkflowParseError("Workflow file root must be a mapping/object.")
+
+ steps = raw.get("steps")
+ if not isinstance(steps, list) or not steps:
+ raise WorkflowParseError(
+ "Workflow file must have a non-empty 'steps' list."
+ )
+
+ seen_names: set = set()
+ validated_steps: List[Dict[str, Any]] = []
+ for index, step in enumerate(steps):
+ validated = _validate_step(step, index, seen_names)
+ substituted = _substitute_value(validated, validated["name"])
+ validated_steps.append(substituted)
+
+ return {
+ "manifest": raw.get("manifest"),
+ "steps": validated_steps,
+ }
\ No newline at end of file
diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py
new file mode 100644
index 0000000..df35065
--- /dev/null
+++ b/databusclient/workflow/steps.py
@@ -0,0 +1,249 @@
+"""Step classes — adapt a workflow step config into a call to the existing
+download()/deploy()/delete() API functions.
+
+Each step class resolves any ${steps.name.key} references in its config via
+StepContext, calls the existing, unmodified API function, and records its
+output back into StepContext so later steps can reference it.
+
+No new business logic lives here. Steps are thin adapters only.
+"""
+
+from __future__ import annotations
+import os
+from typing import Any, Dict
+
+from databusclient.api.delete import delete as api_delete
+from databusclient.api.deploy import (
+ create_dataset,
+ deploy as api_deploy_call,
+ deploy_from_metadata,
+)
+from databusclient.api.download import download as api_download
+from databusclient.extensions import webdav
+from databusclient.manifest.context import ManifestContext
+
+from databusclient.workflow.context import StepContext
+
+
+class StepValidationError(Exception):
+ """Raised when a step's config is missing a required, command-specific field."""
+
+
+class DownloadStep:
+ """Adapts a workflow step to a call to download().
+
+ Accepts either a single URI ('uri') or multiple ('uris') -- the
+ underlying download() function already supports a list.
+
+ output_urls records the ACTUAL, final URL each downloaded file was
+ fetched from -- after any HTTP redirect. download.py's _download_file
+ already resolves redirects internally and reports the final url via
+ manifest_context.record_file(); this step supplies a ManifestContext
+ (the user's real one if set on StepContext, otherwise a throwaway one
+ used purely to capture this information) and reads the resolved URLs
+ back from it, rather than re-deriving redirects itself. Re-deriving
+ was tried first and found to be wrong: a version/artifact/group URI
+ does not redirect the same way an individual file URL does, so
+ checking the input URI directly gives the wrong (un-redirected)
+ answer. Reading what download.py already resolved is correct
+ regardless of whether the input was a single file, version, artifact,
+ or group URI, and regardless of how many files it expanded to.
+ """
+
+ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
+ resolved = context.resolve(step_config)
+ name = resolved["name"]
+
+ uri_value = resolved.get("uri") or resolved.get("uris")
+ if not uri_value:
+ raise StepValidationError(
+ f"Step '{name}': download step requires 'uri' (single) or "
+ f"'uris' (list)."
+ )
+ uris = [uri_value] if isinstance(uri_value, str) else list(uri_value)
+
+ local_dir = resolved.get("localdir")
+ if local_dir is None:
+ local_dir = os.path.join(os.getcwd(), ".workflow", name)
+
+ capture_context = context.manifest_context or ManifestContext(command="download")
+ files_before = len(capture_context.files)
+
+ api_download(
+ localDir=local_dir,
+ endpoint=resolved.get("databus"),
+ databusURIs=uris,
+ token=resolved.get("vault_token"),
+ databus_key=resolved.get("databus_key"),
+ all_versions=resolved.get("all_versions", False),
+ compression=resolved.get("convert_to") or resolved.get("compression"),
+ convert_format=resolved.get("format"),
+ graph_name=resolved.get("graph_name"),
+ base_uri=resolved.get("base_uri"),
+ validate_checksum=resolved.get("validate_checksum", False),
+ manifest_context=capture_context,
+ )
+
+ new_entries = capture_context.files[files_before:]
+ resolved_urls = [e["url"] for e in new_entries if e.get("status") == "success"]
+
+ output_files = self._collect_output_files(local_dir)
+ context.set_output(name, "output_files", output_files)
+ context.set_output(name, "output_urls", resolved_urls)
+
+ @staticmethod
+ def _collect_output_files(local_dir: str) -> list:
+ """Walk local_dir and return all file paths produced by the download.
+
+ Always returns a flat list of file paths, even if the download
+ produced files nested in subdirectories (e.g. a Quad -> Triple
+ split, which writes multiple files into a subdirectory).
+ """
+ if not os.path.isdir(local_dir):
+ return []
+ return sorted(
+ os.path.join(root, filename)
+ for root, _dirs, filenames in os.walk(local_dir)
+ for filename in filenames
+ )
+
+
+class DeployStep:
+ """Adapts a workflow step to a call to create_dataset() + deploy(),
+ or to webdav.upload_to_webdav() + deploy_from_metadata() in WebDAV mode.
+
+ Classic and metadata-file deploy modes operate on their normal inputs
+ (URLs, or an already-resolved metadata list) and must NOT be given
+ local file paths from a previous step -- neither mode can turn a local
+ path into a fetchable URL. Chaining a previous step's local files into
+ a deploy step is only supported via WebDAV mode: local files are
+ uploaded first, which produces real URLs, and any local modifications
+ (format/compression conversions) are correctly reflected since the
+ upload happens after those conversions.
+ """
+
+ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
+ resolved = context.resolve(step_config)
+ name = resolved["name"]
+
+ required = ["version_id", "title", "abstract", "description", "license", "api_key"]
+ missing = [f for f in required if not resolved.get(f)]
+ if missing:
+ raise StepValidationError(
+ f"Step '{name}': deploy step is missing required field(s): "
+ f"{', '.join(missing)}."
+ )
+
+ webdav_url = resolved.get("webdav_url")
+ remote = resolved.get("remote")
+ path = resolved.get("path")
+ webdav_fields = [webdav_url, remote, path]
+
+ if any(webdav_fields) and not all(webdav_fields):
+ raise StepValidationError(
+ f"Step '{name}': WebDAV deploy mode requires 'webdav_url', "
+ f"'remote', and 'path' together."
+ )
+
+ if all(webdav_fields):
+ output_files = self._run_webdav_mode(resolved, name)
+ else:
+ output_files = self._run_classic_mode(resolved, name)
+
+ context.set_output(name, "output_files", output_files)
+ context.set_output(name, "version_id", resolved["version_id"])
+
+ def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list:
+ files = resolved.get("files")
+ if not files:
+ raise StepValidationError(
+ f"Step '{name}': deploy step requires 'files' (a list of URLs)."
+ )
+ if isinstance(files, str):
+ files = [files]
+
+ non_urls = [f for f in files if not str(f).split("|")[0].startswith(("http://", "https://"))]
+ if non_urls:
+ raise StepValidationError(
+ f"Step '{name}': 'files' must be URLs (http:// or https://). "
+ f"Found non-URL value(s): {non_urls}. Local file paths from "
+ f"a previous download step are not accepted in classic "
+ f"deploy mode -- use WebDAV mode ('webdav_url', 'remote', "
+ f"'path') to deploy locally modified files."
+ )
+
+ dataid = create_dataset(
+ version_id=resolved["version_id"],
+ artifact_version_title=resolved["title"],
+ artifact_version_abstract=resolved["abstract"],
+ artifact_version_description=resolved["description"],
+ license_url=resolved["license"],
+ distributions=files,
+ )
+ api_deploy_call(dataid=dataid, api_key=resolved["api_key"])
+ return files
+
+ def _run_webdav_mode(self, resolved: Dict[str, Any], name: str) -> list:
+ local_files = resolved.get("files")
+ if not local_files:
+ raise StepValidationError(
+ f"Step '{name}': WebDAV deploy mode requires 'files' (local "
+ f"file paths to upload, e.g. from a previous download step)."
+ )
+ if isinstance(local_files, str):
+ local_files = [local_files]
+
+ metadata = webdav.upload_to_webdav(
+ local_files, resolved["remote"], resolved["path"], resolved["webdav_url"]
+ )
+ deploy_from_metadata(
+ metadata,
+ resolved["version_id"],
+ resolved["title"],
+ resolved["abstract"],
+ resolved["description"],
+ resolved["license"],
+ resolved["api_key"],
+ )
+ return [entry.get("url", "") for entry in metadata]
+
+
+class DeleteStep:
+ """Adapts a workflow step to a call to delete().
+
+ Workflows are meant to run unattended -- a delete step never triggers
+ the interactive confirmation prompt that the plain `delete` CLI command
+ uses. force is always effectively True here; dry_run must be set
+ explicitly in the step config if a preview-only run is wanted.
+ """
+
+ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
+ resolved = context.resolve(step_config)
+ name = resolved["name"]
+
+ uris = resolved.get("uris")
+ if not uris:
+ raise StepValidationError(f"Step '{name}': delete step requires 'uris'.")
+ if isinstance(uris, str):
+ uris = [uris]
+
+ api_key = resolved.get("api_key")
+ if not api_key:
+ raise StepValidationError(f"Step '{name}': delete step requires 'api_key'.")
+
+ api_delete(
+ databusURIs=uris,
+ databus_key=api_key,
+ dry_run=resolved.get("dry_run", False),
+ force=True,
+ manifest_context=context.manifest_context,
+ )
+
+ context.set_output(name, "output_files", [])
+
+
+STEP_REGISTRY = {
+ "download": DownloadStep,
+ "deploy": DeployStep,
+ "delete": DeleteStep,
+}
\ No newline at end of file
diff --git a/examples/workflows/README.md b/examples/workflows/README.md
new file mode 100644
index 0000000..5b412ec
--- /dev/null
+++ b/examples/workflows/README.md
@@ -0,0 +1,16 @@
+# Example Workflows
+
+Three example workflow pipelines, each runnable directly. All three use real, existing Databus data as their download source.
+
+```bash
+export DATABUS_API_KEY=your-key-here
+databusclient workflow run download-deploy.yml
+```
+
+- **`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.
+
+All three set `api_key: ${DATABUS_API_KEY}` - set that environment variable before running, rather than writing a real key into the 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.
diff --git a/examples/workflows/download-delete.yml b/examples/workflows/download-delete.yml
new file mode 100644
index 0000000..67f47be
--- /dev/null
+++ b/examples/workflows/download-delete.yml
@@ -0,0 +1,11 @@
+steps:
+ - name: archive_dataset
+ command: download
+ uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0
+ localdir: ./workflow-output/archive
+
+ - name: remove_archived_version
+ command: delete
+ uris:
+ - https://databus.dbpedia.org/DhanashreeP/test-group/workflow-demo-deploy/1.0
+ api_key: ${DATABUS_API_KEY}
\ No newline at end of file
diff --git a/examples/workflows/download-deploy.yml b/examples/workflows/download-deploy.yml
new file mode 100644
index 0000000..0b0816a
--- /dev/null
+++ b/examples/workflows/download-deploy.yml
@@ -0,0 +1,16 @@
+steps:
+ - name: fetch_dataset
+ command: download
+ uri: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-source-data/1.0
+ localdir: ./workflow-output/download-deploy
+
+ - name: publish_dataset
+ command: deploy
+ version_id: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-demo-deploy/1.0
+ title: "Workflow Demo - Download and Deploy"
+ abstract: "throwaway, testing workflow deploy chaining"
+ description: "throwaway version testing deploy chaining via output_urls"
+ license: https://creativecommons.org/licenses/by-sa/3.0/
+ api_key: ${DATABUS_API_KEY}
+ files: ${steps.fetch_dataset.output_urls}
+ on_error: fail
\ No newline at end of file
diff --git a/examples/workflows/full-pipeline.yml b/examples/workflows/full-pipeline.yml
new file mode 100644
index 0000000..4ec618c
--- /dev/null
+++ b/examples/workflows/full-pipeline.yml
@@ -0,0 +1,23 @@
+steps:
+ - name: fetch_dataset
+ command: download
+ uri: https://databus.dev.dbpedia.link/fhofer/gsoc26/test-data/2.0
+ localdir: ./workflow-output/full-pipeline
+
+ - name: publish_dataset
+ command: deploy
+ version_id: https://databus.dbpedia.org/DhanashreeP/test-group/workflow-full-pipeline-demo/1.0
+ title: "Workflow Demo - Full Pipeline"
+ abstract: "throwaway, testing full download-deploy-delete pipeline"
+ description: "throwaway version testing all three commands chained together, using real Databus test data as source"
+ license: https://creativecommons.org/licenses/by-sa/3.0/
+ api_key: ${DATABUS_API_KEY}
+ files: ${steps.fetch_dataset.output_urls}
+ on_error: fail
+
+ - name: cleanup_previous_version
+ command: delete
+ uris:
+ - https://databus.dbpedia.org/DhanashreeP/test-group/workflow-full-pipeline-demo/1.0
+ api_key: ${DATABUS_API_KEY}
+ on_error: continue
\ No newline at end of file
diff --git a/poetry.lock b/poetry.lock
index e3759ff..88e0a8e 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand.
[[package]]
name = "black"
@@ -329,6 +329,89 @@ pluggy = ">=0.12,<2.0"
[package.extras]
testing = ["argcomplete", "attrs (>=19.2.0)", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
+[[package]]
+name = "pyyaml"
+version = "6.0.3"
+description = "YAML parser and emitter for Python"
+optional = false
+python-versions = ">=3.8"
+groups = ["main"]
+files = [
+ {file = "PyYAML-6.0.3-cp38-cp38-macosx_10_13_x86_64.whl", hash = "sha256:c2514fceb77bc5e7a2f7adfaa1feb2fb311607c9cb518dbc378688ec73d8292f"},
+ {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9c57bb8c96f6d1808c030b1687b9b5fb476abaa47f0db9c0101f5e9f394e97f4"},
+ {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:efd7b85f94a6f21e4932043973a7ba2613b059c4a000551892ac9f1d11f5baf3"},
+ {file = "PyYAML-6.0.3-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:22ba7cfcad58ef3ecddc7ed1db3409af68d023b7f940da23c6c2a1890976eda6"},
+ {file = "PyYAML-6.0.3-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:6344df0d5755a2c9a276d4473ae6b90647e216ab4757f8426893b5dd2ac3f369"},
+ {file = "PyYAML-6.0.3-cp38-cp38-win32.whl", hash = "sha256:3ff07ec89bae51176c0549bc4c63aa6202991da2d9a6129d7aef7f1407d3f295"},
+ {file = "PyYAML-6.0.3-cp38-cp38-win_amd64.whl", hash = "sha256:5cf4e27da7e3fbed4d6c3d8e797387aaad68102272f8f9752883bc32d61cb87b"},
+ {file = "pyyaml-6.0.3-cp310-cp310-macosx_10_13_x86_64.whl", hash = "sha256:214ed4befebe12df36bcc8bc2b64b396ca31be9304b8f59e25c11cf94a4c033b"},
+ {file = "pyyaml-6.0.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:02ea2dfa234451bbb8772601d7b8e426c2bfa197136796224e50e35a78777956"},
+ {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b30236e45cf30d2b8e7b3e85881719e98507abed1011bf463a8fa23e9c3e98a8"},
+ {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:66291b10affd76d76f54fad28e22e51719ef9ba22b29e1d7d03d6777a9174198"},
+ {file = "pyyaml-6.0.3-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9c7708761fccb9397fe64bbc0395abcae8c4bf7b0eac081e12b809bf47700d0b"},
+ {file = "pyyaml-6.0.3-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:418cf3f2111bc80e0933b2cd8cd04f286338bb88bdc7bc8e6dd775ebde60b5e0"},
+ {file = "pyyaml-6.0.3-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:5e0b74767e5f8c593e8c9b5912019159ed0533c70051e9cce3e8b6aa699fcd69"},
+ {file = "pyyaml-6.0.3-cp310-cp310-win32.whl", hash = "sha256:28c8d926f98f432f88adc23edf2e6d4921ac26fb084b028c733d01868d19007e"},
+ {file = "pyyaml-6.0.3-cp310-cp310-win_amd64.whl", hash = "sha256:bdb2c67c6c1390b63c6ff89f210c8fd09d9a1217a465701eac7316313c915e4c"},
+ {file = "pyyaml-6.0.3-cp311-cp311-macosx_10_13_x86_64.whl", hash = "sha256:44edc647873928551a01e7a563d7452ccdebee747728c1080d881d68af7b997e"},
+ {file = "pyyaml-6.0.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:652cb6edd41e718550aad172851962662ff2681490a8a711af6a4d288dd96824"},
+ {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:10892704fc220243f5305762e276552a0395f7beb4dbf9b14ec8fd43b57f126c"},
+ {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:850774a7879607d3a6f50d36d04f00ee69e7fc816450e5f7e58d7f17f1ae5c00"},
+ {file = "pyyaml-6.0.3-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b8bb0864c5a28024fac8a632c443c87c5aa6f215c0b126c449ae1a150412f31d"},
+ {file = "pyyaml-6.0.3-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1d37d57ad971609cf3c53ba6a7e365e40660e3be0e5175fa9f2365a379d6095a"},
+ {file = "pyyaml-6.0.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:37503bfbfc9d2c40b344d06b2199cf0e96e97957ab1c1b546fd4f87e53e5d3e4"},
+ {file = "pyyaml-6.0.3-cp311-cp311-win32.whl", hash = "sha256:8098f252adfa6c80ab48096053f512f2321f0b998f98150cea9bd23d83e1467b"},
+ {file = "pyyaml-6.0.3-cp311-cp311-win_amd64.whl", hash = "sha256:9f3bfb4965eb874431221a3ff3fdcddc7e74e3b07799e0e84ca4a0f867d449bf"},
+ {file = "pyyaml-6.0.3-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7f047e29dcae44602496db43be01ad42fc6f1cc0d8cd6c83d342306c32270196"},
+ {file = "pyyaml-6.0.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:fc09d0aa354569bc501d4e787133afc08552722d3ab34836a80547331bb5d4a0"},
+ {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9149cad251584d5fb4981be1ecde53a1ca46c891a79788c0df828d2f166bda28"},
+ {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:5fdec68f91a0c6739b380c83b951e2c72ac0197ace422360e6d5a959d8d97b2c"},
+ {file = "pyyaml-6.0.3-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ba1cc08a7ccde2d2ec775841541641e4548226580ab850948cbfda66a1befcdc"},
+ {file = "pyyaml-6.0.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:8dc52c23056b9ddd46818a57b78404882310fb473d63f17b07d5c40421e47f8e"},
+ {file = "pyyaml-6.0.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:41715c910c881bc081f1e8872880d3c650acf13dfa8214bad49ed4cede7c34ea"},
+ {file = "pyyaml-6.0.3-cp312-cp312-win32.whl", hash = "sha256:96b533f0e99f6579b3d4d4995707cf36df9100d67e0c8303a0c55b27b5f99bc5"},
+ {file = "pyyaml-6.0.3-cp312-cp312-win_amd64.whl", hash = "sha256:5fcd34e47f6e0b794d17de1b4ff496c00986e1c83f7ab2fb8fcfe9616ff7477b"},
+ {file = "pyyaml-6.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:64386e5e707d03a7e172c0701abfb7e10f0fb753ee1d773128192742712a98fd"},
+ {file = "pyyaml-6.0.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8da9669d359f02c0b91ccc01cac4a67f16afec0dac22c2ad09f46bee0697eba8"},
+ {file = "pyyaml-6.0.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2283a07e2c21a2aa78d9c4442724ec1eb15f5e42a723b99cb3d822d48f5f7ad1"},
+ {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ee2922902c45ae8ccada2c5b501ab86c36525b883eff4255313a253a3160861c"},
+ {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a33284e20b78bd4a18c8c2282d549d10bc8408a2a7ff57653c0cf0b9be0afce5"},
+ {file = "pyyaml-6.0.3-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0f29edc409a6392443abf94b9cf89ce99889a1dd5376d94316ae5145dfedd5d6"},
+ {file = "pyyaml-6.0.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:f7057c9a337546edc7973c0d3ba84ddcdf0daa14533c2065749c9075001090e6"},
+ {file = "pyyaml-6.0.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:eda16858a3cab07b80edaf74336ece1f986ba330fdb8ee0d6c0d68fe82bc96be"},
+ {file = "pyyaml-6.0.3-cp313-cp313-win32.whl", hash = "sha256:d0eae10f8159e8fdad514efdc92d74fd8d682c933a6dd088030f3834bc8e6b26"},
+ {file = "pyyaml-6.0.3-cp313-cp313-win_amd64.whl", hash = "sha256:79005a0d97d5ddabfeeea4cf676af11e647e41d81c9a7722a193022accdb6b7c"},
+ {file = "pyyaml-6.0.3-cp313-cp313-win_arm64.whl", hash = "sha256:5498cd1645aa724a7c71c8f378eb29ebe23da2fc0d7a08071d89469bf1d2defb"},
+ {file = "pyyaml-6.0.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:8d1fab6bb153a416f9aeb4b8763bc0f22a5586065f86f7664fc23339fc1c1fac"},
+ {file = "pyyaml-6.0.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:34d5fcd24b8445fadc33f9cf348c1047101756fd760b4dacb5c3e99755703310"},
+ {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:501a031947e3a9025ed4405a168e6ef5ae3126c59f90ce0cd6f2bfc477be31b7"},
+ {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b3bc83488de33889877a0f2543ade9f70c67d66d9ebb4ac959502e12de895788"},
+ {file = "pyyaml-6.0.3-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c458b6d084f9b935061bc36216e8a69a7e293a2f1e68bf956dcd9e6cbcd143f5"},
+ {file = "pyyaml-6.0.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7c6610def4f163542a622a73fb39f534f8c101d690126992300bf3207eab9764"},
+ {file = "pyyaml-6.0.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5190d403f121660ce8d1d2c1bb2ef1bd05b5f68533fc5c2ea899bd15f4399b35"},
+ {file = "pyyaml-6.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:4a2e8cebe2ff6ab7d1050ecd59c25d4c8bd7e6f400f5f82b96557ac0abafd0ac"},
+ {file = "pyyaml-6.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:93dda82c9c22deb0a405ea4dc5f2d0cda384168e466364dec6255b293923b2f3"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:02893d100e99e03eda1c8fd5c441d8c60103fd175728e23e431db1b589cf5ab3"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:c1ff362665ae507275af2853520967820d9124984e0f7466736aea23d8611fba"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6adc77889b628398debc7b65c073bcb99c4a0237b248cacaf3fe8a557563ef6c"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a80cb027f6b349846a3bf6d73b5e95e782175e52f22108cfa17876aaeff93702"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:00c4bdeba853cc34e7dd471f16b4114f4162dc03e6b7afcc2128711f0eca823c"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:66e1674c3ef6f541c35191caae2d429b967b99e02040f5ba928632d9a7f0f065"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:16249ee61e95f858e83976573de0f5b2893b3677ba71c9dd36b9cf8be9ac6d65"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:4ad1906908f2f5ae4e5a8ddfce73c320c2a1429ec52eafd27138b7f1cbe341c9"},
+ {file = "pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b"},
+ {file = "pyyaml-6.0.3-cp39-cp39-macosx_10_13_x86_64.whl", hash = "sha256:b865addae83924361678b652338317d1bd7e79b1f4596f96b96c77a5a34b34da"},
+ {file = "pyyaml-6.0.3-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:c3355370a2c156cffb25e876646f149d5d68f5e0a3ce86a5084dd0b64a994917"},
+ {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3c5677e12444c15717b902a5798264fa7909e41153cdf9ef7ad571b704a63dd9"},
+ {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:5ed875a24292240029e4483f9d4a4b8a1ae08843b9c54f43fcc11e404532a8a5"},
+ {file = "pyyaml-6.0.3-cp39-cp39-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0150219816b6a1fa26fb4699fb7daa9caf09eb1999f3b70fb6e786805e80375a"},
+ {file = "pyyaml-6.0.3-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:fa160448684b4e94d80416c0fa4aac48967a969efe22931448d853ada8baf926"},
+ {file = "pyyaml-6.0.3-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:27c0abcb4a5dac13684a37f76e701e054692a9b2d3064b70f5e4eb54810553d7"},
+ {file = "pyyaml-6.0.3-cp39-cp39-win32.whl", hash = "sha256:1ebe39cb5fc479422b83de611d14e2c0d3bb2a18bbcb01f229ab3cfbd8fee7a0"},
+ {file = "pyyaml-6.0.3-cp39-cp39-win_amd64.whl", hash = "sha256:2e71d11abed7344e42a8849600193d15b6def118602c4c176f748e4583246007"},
+ {file = "pyyaml-6.0.3.tar.gz", hash = "sha256:d76623373421df22fb4cf8817020cbb7ef15c725b9d5e45f17e189bfc384190f"},
+]
+
[[package]]
name = "rdflib"
version = "7.5.0"
@@ -466,4 +549,4 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""]
[metadata]
lock-version = "2.1"
python-versions = "^3.11"
-content-hash = "f625db7ea6714ebf87336efecaef03ec2dc4f6f7838c3239432828cd6649ff96"
+content-hash = "b738c415f513b772068e55993bbaf06c4b8db37a77f5303e21b9498636b7c91b"
diff --git a/pyproject.toml b/pyproject.toml
index e1485ae..d1cdc47 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -13,6 +13,7 @@ requests = "^2.28.1"
tqdm = "^4.42.1"
SPARQLWrapper = "^2.0.0"
rdflib = "^7.2.1"
+pyyaml = "^6.0.3"
[tool.poetry.group.dev.dependencies]
black = "^22.6.0"
diff --git a/tests/test_step_context.py b/tests/test_step_context.py
new file mode 100644
index 0000000..a90f62e
--- /dev/null
+++ b/tests/test_step_context.py
@@ -0,0 +1,75 @@
+"""Tests for StepContext (Milestone 4)."""
+
+import pytest
+
+from databusclient.workflow.context import StepContext, StepReferenceError
+
+
+def test_set_and_get_output():
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["/data/a.ttl", "/data/b.ttl"])
+ assert ctx.get_output("fetch", "output_files") == ["/data/a.ttl", "/data/b.ttl"]
+
+
+def test_get_output_unknown_step_raises():
+ ctx = StepContext()
+ with pytest.raises(StepReferenceError, match="unknown or not-yet-executed"):
+ ctx.get_output("nope", "output_files")
+
+
+def test_get_output_unknown_key_raises():
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["/data/a.ttl"])
+ with pytest.raises(StepReferenceError, match="no recorded output"):
+ ctx.get_output("fetch", "some_other_key")
+
+
+def test_resolve_exact_token_preserves_list_type():
+ """A value that IS exactly one ${steps.x.y} token resolves to the raw list."""
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["/data/a.ttl", "/data/b.ttl"])
+ resolved = ctx.resolve("${steps.fetch.output_files}")
+ assert resolved == ["/data/a.ttl", "/data/b.ttl"]
+ assert isinstance(resolved, list)
+
+
+def test_resolve_embedded_token_in_string():
+ ctx = StepContext()
+ ctx.set_output("fetch", "version", "2024.01")
+ resolved = ctx.resolve("Deployed version ${steps.fetch.version}")
+ assert resolved == "Deployed version 2024.01"
+
+
+def test_resolve_nested_dict_and_list():
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["/data/a.ttl"])
+ resolved = ctx.resolve({
+ "files": "${steps.fetch.output_files}",
+ "meta": {"note": "from ${steps.fetch.output_files}"},
+ })
+ assert resolved["files"] == ["/data/a.ttl"]
+ assert resolved["meta"]["note"] == "from ['/data/a.ttl']"
+
+
+def test_resolve_plain_value_passthrough():
+ ctx = StepContext()
+ assert ctx.resolve("no tokens here") == "no tokens here"
+ assert ctx.resolve(42) == 42
+ assert ctx.resolve(None) is None
+
+
+def test_resolve_unresolvable_reference_raises():
+ ctx = StepContext()
+ with pytest.raises(StepReferenceError):
+ ctx.resolve("${steps.never_ran.output_files}")
+
+
+def test_manifest_context_defaults_to_none():
+ ctx = StepContext()
+ assert ctx.manifest_context is None
+
+
+def test_manifest_context_stored_when_provided():
+ sentinel = object()
+ ctx = StepContext(manifest_context=sentinel)
+ assert ctx.manifest_context is sentinel
\ No newline at end of file
diff --git a/tests/test_workflow_engine.py b/tests/test_workflow_engine.py
new file mode 100644
index 0000000..71005a4
--- /dev/null
+++ b/tests/test_workflow_engine.py
@@ -0,0 +1,136 @@
+"""Tests for WorkflowEngine (Milestone 4)."""
+
+import pytest
+
+from databusclient.workflow.engine import WorkflowEngine, WorkflowExecutionError
+
+def test_runs_steps_in_order(monkeypatch):
+ order = []
+
+ class OrderedStep:
+ def run(self, step_config, context):
+ order.append(step_config["name"])
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", OrderedStep)
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", OrderedStep)
+
+ engine = WorkflowEngine()
+ engine.run([
+ {"name": "a", "command": "download"},
+ {"name": "b", "command": "deploy"},
+ ])
+ assert order == ["a", "b"]
+
+
+def test_step_failure_with_default_fail_raises(monkeypatch):
+ class FailingStep:
+ def run(self, step_config, context):
+ raise RuntimeError("boom")
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FailingStep)
+
+ engine = WorkflowEngine()
+ with pytest.raises(WorkflowExecutionError, match="boom"):
+ engine.run([{"name": "a", "command": "download"}])
+
+
+def test_step_failure_with_continue_does_not_raise(monkeypatch):
+ class FailingStep:
+ def run(self, step_config, context):
+ raise RuntimeError("boom")
+
+ class OKStep:
+ def run(self, step_config, context):
+ pass
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FailingStep)
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", OKStep)
+
+ engine = WorkflowEngine()
+ results = engine.run([
+ {"name": "a", "command": "download", "on_error": "continue"},
+ {"name": "b", "command": "deploy"},
+ ])
+ assert results[0].status == "skipped_error"
+ assert results[1].status == "success"
+
+
+def test_retry_succeeds_on_second_attempt(monkeypatch):
+ attempts = {"count": 0}
+
+ class FlakyStep:
+ def run(self, step_config, context):
+ attempts["count"] += 1
+ if attempts["count"] < 2:
+ raise RuntimeError("transient failure")
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FlakyStep)
+
+ engine = WorkflowEngine()
+ results = engine.run([{
+ "name": "a", "command": "download", "on_error": "retry",
+ "retry": {"max_attempts": 3, "delay_seconds": 0},
+ }])
+ assert results[0].status == "success"
+ assert results[0].attempts == 2
+ assert attempts["count"] == 2
+
+
+def test_retry_exhausts_attempts_and_fails(monkeypatch):
+ class AlwaysFailsStep:
+ def run(self, step_config, context):
+ raise RuntimeError("permanent failure")
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", AlwaysFailsStep)
+
+ engine = WorkflowEngine()
+ with pytest.raises(WorkflowExecutionError, match="permanent failure"):
+ engine.run([{
+ "name": "a", "command": "download", "on_error": "retry",
+ "retry": {"max_attempts": 2, "delay_seconds": 0},
+ }])
+
+
+def test_step_chaining_end_to_end(monkeypatch):
+ """A download step's output is available to a deploy step via StepContext."""
+ class FetchStep:
+ def run(self, step_config, context):
+ context.set_output(step_config["name"], "output_files", ["/data/a.ttl"])
+
+ captured = {}
+
+ class PublishStep:
+ def run(self, step_config, context):
+ resolved = context.resolve(step_config)
+ captured["files"] = resolved["files"]
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "download", FetchStep)
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", PublishStep)
+
+ engine = WorkflowEngine()
+ engine.run([
+ {"name": "fetch", "command": "download"},
+ {"name": "publish", "command": "deploy", "files": "${steps.fetch.output_files}"},
+ ])
+ assert captured["files"] == ["/data/a.ttl"]
+
+def test_unknown_step_reference_surfaces_as_workflow_execution_error(monkeypatch):
+ class PublishStep:
+ def run(self, step_config, context):
+ context.resolve(step_config) # will raise, since "nonexistent" never ran
+
+ from databusclient.workflow import steps as steps_module
+ monkeypatch.setitem(steps_module.STEP_REGISTRY, "deploy", PublishStep)
+
+ engine = WorkflowEngine()
+ with pytest.raises(WorkflowExecutionError, match="unknown or not-yet-executed"):
+ engine.run([
+ {"name": "publish", "command": "deploy",
+ "files": "${steps.nonexistent.output_files}"},
+ ])
\ No newline at end of file
diff --git a/tests/test_workflow_parser.py b/tests/test_workflow_parser.py
new file mode 100644
index 0000000..b82bab7
--- /dev/null
+++ b/tests/test_workflow_parser.py
@@ -0,0 +1,180 @@
+"""Tests for WorkflowParser (Milestone 4)."""
+
+import os
+import tempfile
+
+import pytest
+import yaml
+
+from databusclient.workflow.parser import (
+ MissingEnvVarError,
+ WorkflowParseError,
+ parse_workflow,
+)
+
+
+def _write_yaml(content: dict) -> str:
+ fd, path = tempfile.mkstemp(suffix=".yml")
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
+ yaml.safe_dump(content, f)
+ return path
+
+
+def test_parses_minimal_valid_workflow():
+ path = _write_yaml({
+ "steps": [
+ {"name": "fetch", "command": "download", "uri": "https://example.org/x"},
+ ]
+ })
+ result = parse_workflow(path)
+ assert result["manifest"] is None
+ assert len(result["steps"]) == 1
+ assert result["steps"][0]["name"] == "fetch"
+
+
+def test_missing_steps_key_raises():
+ path = _write_yaml({"manifest": "run.json"})
+ with pytest.raises(WorkflowParseError, match="steps"):
+ parse_workflow(path)
+
+
+def test_empty_steps_list_raises():
+ path = _write_yaml({"steps": []})
+ with pytest.raises(WorkflowParseError, match="non-empty"):
+ parse_workflow(path)
+
+
+def test_step_missing_name_raises():
+ path = _write_yaml({"steps": [{"command": "download", "uri": "x"}]})
+ with pytest.raises(WorkflowParseError, match="name"):
+ parse_workflow(path)
+
+
+def test_duplicate_step_names_raise():
+ path = _write_yaml({
+ "steps": [
+ {"name": "a", "command": "download", "uri": "x"},
+ {"name": "a", "command": "delete", "uris": ["x"]},
+ ]
+ })
+ with pytest.raises(WorkflowParseError, match="Duplicate step name"):
+ parse_workflow(path)
+
+
+def test_invalid_command_raises():
+ path = _write_yaml({"steps": [{"name": "a", "command": "bogus"}]})
+ with pytest.raises(WorkflowParseError, match="invalid command"):
+ parse_workflow(path)
+
+
+def test_invalid_on_error_raises():
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "download", "uri": "x", "on_error": "maybe"}]
+ })
+ with pytest.raises(WorkflowParseError, match="invalid on_error"):
+ parse_workflow(path)
+
+
+def test_retry_without_config_raises():
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "download", "uri": "x", "on_error": "retry"}]
+ })
+ with pytest.raises(WorkflowParseError, match="retry"):
+ parse_workflow(path)
+
+
+def test_retry_with_invalid_max_attempts_raises():
+ path = _write_yaml({
+ "steps": [{
+ "name": "a", "command": "download", "uri": "x", "on_error": "retry",
+ "retry": {"max_attempts": 0, "delay_seconds": 5},
+ }]
+ })
+ with pytest.raises(WorkflowParseError, match="max_attempts"):
+ parse_workflow(path)
+
+
+def test_valid_retry_config_passes(monkeypatch):
+ path = _write_yaml({
+ "steps": [{
+ "name": "a", "command": "download", "uri": "x", "on_error": "retry",
+ "retry": {"max_attempts": 3, "delay_seconds": 5},
+ }]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][0]["retry"]["max_attempts"] == 3
+
+
+def test_env_var_substitution(monkeypatch):
+ monkeypatch.setenv("MY_API_KEY", "secret123")
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "deploy", "api_key": "${MY_API_KEY}"}]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][0]["api_key"] == "secret123"
+
+
+def test_missing_env_var_raises(monkeypatch):
+ monkeypatch.delenv("DOES_NOT_EXIST_VAR", raising=False)
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "deploy", "api_key": "${DOES_NOT_EXIST_VAR}"}]
+ })
+ with pytest.raises(MissingEnvVarError, match="DOES_NOT_EXIST_VAR"):
+ parse_workflow(path)
+
+
+def test_steps_reference_left_untouched():
+ """${steps.x.output_files} must NOT be treated as a missing env var."""
+ path = _write_yaml({
+ "steps": [
+ {"name": "fetch", "command": "download", "uri": "x"},
+ {"name": "publish", "command": "deploy", "files": "${steps.fetch.output_files}"},
+ ]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][1]["files"] == "${steps.fetch.output_files}"
+
+
+def test_multiple_tokens_in_same_string(monkeypatch):
+ monkeypatch.setenv("ACCOUNT", "myaccount")
+ monkeypatch.setenv("GROUP", "mygroup")
+ path = _write_yaml({
+ "steps": [{
+ "name": "a", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/${ACCOUNT}/${GROUP}/art/1.0",
+ }]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][0]["version_id"] == "https://databus.dbpedia.org/myaccount/mygroup/art/1.0"
+
+
+def test_nonexistent_file_raises():
+ with pytest.raises(WorkflowParseError, match="not found"):
+ parse_workflow("does-not-exist.yml")
+
+
+def test_invalid_yaml_raises():
+ fd, path = tempfile.mkstemp(suffix=".yml")
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
+ f.write("steps: [unclosed")
+ with pytest.raises(WorkflowParseError, match="not valid YAML"):
+ parse_workflow(path)
+
+def test_bare_dollar_var_without_braces_passes_through_unchanged():
+ """$VAR (no braces) is not a substitution token -- left as literal text."""
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "download", "uri": "$HOME/data"}]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][0]["uri"] == "$HOME/data"
+
+
+def test_empty_braces_pass_through_unchanged():
+ """${} has no characters between the braces, so it doesn't match the
+ substitution pattern at all (which requires at least one character) --
+ it passes through as literal text, same as a bare $VAR without braces."""
+ path = _write_yaml({
+ "steps": [{"name": "a", "command": "download", "uri": "${}/data"}]
+ })
+ result = parse_workflow(path)
+ assert result["steps"][0]["uri"] == "${}/data"
\ No newline at end of file
diff --git a/tests/test_workflow_steps.py b/tests/test_workflow_steps.py
new file mode 100644
index 0000000..d89c0e5
--- /dev/null
+++ b/tests/test_workflow_steps.py
@@ -0,0 +1,373 @@
+"""Tests for step classes (Milestone 4). No live Databus calls -- the
+underlying api_download/api_deploy_call/api_delete functions are mocked."""
+
+import os
+import pytest
+
+from databusclient.manifest.context import ManifestContext
+from databusclient.workflow.context import StepContext
+from databusclient.workflow.steps import (
+ DeleteStep,
+ DeployStep,
+ DownloadStep,
+ StepValidationError,
+)
+
+
+def test_download_step_requires_uri():
+ ctx = StepContext()
+ step = DownloadStep()
+ with pytest.raises(StepValidationError, match="requires 'uri'"):
+ step.run({"name": "fetch", "command": "download"}, ctx)
+
+
+def test_download_step_calls_api_download_and_collects_files(monkeypatch, tmp_path):
+ captured = {}
+
+ def fake_download(**kwargs):
+ captured.update(kwargs)
+ local_dir = kwargs["localDir"]
+ os.makedirs(local_dir, exist_ok=True)
+ with open(os.path.join(local_dir, "a.ttl"), "w") as f:
+ f.write("data")
+ kwargs["manifest_context"].record_file(
+ url=kwargs["databusURIs"][0], status="success"
+ )
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ ctx = StepContext()
+ step = DownloadStep()
+ step.run(
+ {"name": "fetch", "command": "download", "uri": "https://example.org/x",
+ "localdir": str(tmp_path)},
+ ctx,
+ )
+
+ assert captured["databusURIs"] == ["https://example.org/x"]
+ output = ctx.get_output("fetch", "output_files")
+ assert len(output) == 1
+ assert output[0].endswith("a.ttl")
+
+
+def test_download_step_collects_files_from_subdirectory(monkeypatch, tmp_path):
+ """Simulates a Quad -> Triple split producing files in a subdirectory."""
+ def fake_download(**kwargs):
+ local_dir = kwargs["localDir"]
+ sub = os.path.join(local_dir, "split")
+ os.makedirs(sub, exist_ok=True)
+ with open(os.path.join(sub, "graph1.nt"), "w") as f:
+ f.write("data")
+ with open(os.path.join(sub, "graph2.nt"), "w") as f:
+ f.write("data")
+ kwargs["manifest_context"].record_file(
+ url=kwargs["databusURIs"][0], status="success"
+ )
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ ctx = StepContext()
+ step = DownloadStep()
+ step.run(
+ {"name": "fetch", "command": "download", "uri": "x", "localdir": str(tmp_path)},
+ ctx,
+ )
+ output = ctx.get_output("fetch", "output_files")
+ assert len(output) == 2
+ assert all(isinstance(p, str) for p in output)
+
+
+def test_download_step_records_output_urls_from_manifest_context(monkeypatch, tmp_path):
+ """output_urls comes from what download.py itself resolved and recorded
+ -- not from re-checking the input URI, so this test uses an input URI
+ that DIFFERS from the resolved one, exactly like a real Databus
+ redirect would produce."""
+ def fake_download(**kwargs):
+ local_dir = kwargs["localDir"]
+ os.makedirs(local_dir, exist_ok=True)
+ with open(os.path.join(local_dir, "a.ttl"), "w") as f:
+ f.write("data")
+ # Simulates download.py resolving a redirect: the recorded url
+ # differs from the input databusURIs[0].
+ kwargs["manifest_context"].record_file(
+ url="https://raw.githubusercontent.com/real/a.ttl", status="success"
+ )
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ ctx = StepContext()
+ step = DownloadStep()
+ step.run(
+ {"name": "fetch", "command": "download",
+ "uri": "https://databus.dbpedia.org/acct/grp/art/1.0/a.ttl",
+ "localdir": str(tmp_path)},
+ ctx,
+ )
+
+ assert ctx.get_output("fetch", "output_urls") == [
+ "https://raw.githubusercontent.com/real/a.ttl"
+ ]
+
+
+def test_download_step_output_urls_handles_multiple_files(monkeypatch, tmp_path):
+ """A version/artifact/group download can produce multiple files --
+ output_urls must contain the resolved URL for each one."""
+ def fake_download(**kwargs):
+ local_dir = kwargs["localDir"]
+ os.makedirs(local_dir, exist_ok=True)
+ for name in ("a.ttl", "b.ttl"):
+ with open(os.path.join(local_dir, name), "w") as f:
+ f.write("data")
+ ctx = kwargs["manifest_context"]
+ ctx.record_file(url="https://real.example.org/a.ttl", status="success")
+ ctx.record_file(url="https://real.example.org/b.ttl", status="success")
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ ctx = StepContext()
+ step = DownloadStep()
+ step.run(
+ {"name": "fetch", "command": "download",
+ "uri": "https://databus.dbpedia.org/acct/grp/art/1.0",
+ "localdir": str(tmp_path)},
+ ctx,
+ )
+
+ assert ctx.get_output("fetch", "output_urls") == [
+ "https://real.example.org/a.ttl", "https://real.example.org/b.ttl"
+ ]
+
+
+def test_download_step_accepts_multiple_uris(monkeypatch, tmp_path):
+ captured = {}
+
+ def fake_download(**kwargs):
+ captured.update(kwargs)
+ local_dir = kwargs["localDir"]
+ os.makedirs(local_dir, exist_ok=True)
+ with open(os.path.join(local_dir, "a.ttl"), "w") as f:
+ f.write("data")
+ for uri in kwargs["databusURIs"]:
+ kwargs["manifest_context"].record_file(url=uri, status="success")
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ ctx = StepContext()
+ step = DownloadStep()
+ step.run({
+ "name": "fetch", "command": "download",
+ "uris": ["https://example.org/a", "https://example.org/b"],
+ "localdir": str(tmp_path),
+ }, ctx)
+
+ assert captured["databusURIs"] == ["https://example.org/a", "https://example.org/b"]
+ assert ctx.get_output("fetch", "output_urls") == [
+ "https://example.org/a", "https://example.org/b"
+ ]
+
+
+def test_download_step_only_captures_entries_from_this_run(monkeypatch, tmp_path):
+ """If a real, shared manifest_context is used (future Milestone 5),
+ entries from a PRIOR step must not leak into this step's output_urls."""
+ def fake_download(**kwargs):
+ local_dir = kwargs["localDir"]
+ os.makedirs(local_dir, exist_ok=True)
+ with open(os.path.join(local_dir, "b.ttl"), "w") as f:
+ f.write("data")
+ kwargs["manifest_context"].record_file(
+ url="https://example.org/b.ttl", status="success"
+ )
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+ shared_context = ManifestContext(command="download")
+ shared_context.record_file(url="https://example.org/PRIOR.ttl", status="success")
+
+ ctx = StepContext(manifest_context=shared_context)
+ step = DownloadStep()
+ step.run(
+ {"name": "fetch", "command": "download", "uri": "x", "localdir": str(tmp_path)},
+ ctx,
+ )
+
+ assert ctx.get_output("fetch", "output_urls") == ["https://example.org/b.ttl"]
+
+def test_deploy_step_requires_fields():
+ ctx = StepContext()
+ step = DeployStep()
+ with pytest.raises(StepValidationError, match="missing required field"):
+ step.run({"name": "publish", "command": "deploy"}, ctx)
+
+
+def test_deploy_step_resolves_step_reference_to_urls_and_calls_deploy(monkeypatch):
+ """Chaining a step reference into classic mode works when the referenced
+ output is itself URLs (e.g. output_urls from a download step) -- not
+ local file paths. See test_deploy_step_classic_mode_rejects_local_paths_
+ with_clear_error for the local-path rejection case."""
+ captured = {}
+
+ def fake_create_dataset(**kwargs):
+ captured["create_dataset_kwargs"] = kwargs
+ return {"@graph": [{"@id": "fake"}]}
+
+ def fake_deploy(dataid, api_key):
+ captured["api_key"] = api_key
+
+ monkeypatch.setattr("databusclient.workflow.steps.create_dataset", fake_create_dataset)
+ monkeypatch.setattr("databusclient.workflow.steps.api_deploy_call", fake_deploy)
+
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_urls", ["https://example.org/data/a.ttl"])
+
+ step = DeployStep()
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ "files": "${steps.fetch.output_urls}",
+ }, ctx)
+
+ assert captured["create_dataset_kwargs"]["distributions"] == ["https://example.org/data/a.ttl"]
+ assert captured["api_key"] == "key123"
+
+
+def test_delete_step_always_forces_no_prompt(monkeypatch):
+ captured = {}
+
+ def fake_delete(**kwargs):
+ captured.update(kwargs)
+
+ monkeypatch.setattr("databusclient.workflow.steps.api_delete", fake_delete)
+
+ ctx = StepContext()
+ step = DeleteStep()
+ step.run({
+ "name": "cleanup", "command": "delete",
+ "uris": ["https://databus.dbpedia.org/a/b/c/old"],
+ "api_key": "key123",
+ }, ctx)
+
+ assert captured["force"] is True
+ assert captured["dry_run"] is False
+
+
+def test_delete_step_requires_uris():
+ ctx = StepContext()
+ step = DeleteStep()
+ with pytest.raises(StepValidationError, match="requires 'uris'"):
+ step.run({"name": "cleanup", "command": "delete", "api_key": "k"}, ctx)
+
+
+def test_delete_step_requires_api_key():
+ ctx = StepContext()
+ step = DeleteStep()
+ with pytest.raises(StepValidationError, match="requires 'api_key'"):
+ step.run({"name": "cleanup", "command": "delete", "uris": ["x"]}, ctx)
+
+
+def test_deploy_step_classic_mode_rejects_missing_files():
+ ctx = StepContext()
+ step = DeployStep()
+ with pytest.raises(StepValidationError, match="requires 'files'"):
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ }, ctx)
+
+
+def test_deploy_step_classic_mode_rejects_local_paths_with_clear_error():
+ """The actual bug we hit manually: classic mode given local file paths
+ (e.g. chained from a download step's output_files) must fail with a
+ clear, actionable error -- not a raw 'Invalid URL' crash."""
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["./tmp/workflow-demo/download/swagger.yml"])
+
+ step = DeployStep()
+ with pytest.raises(StepValidationError, match="Local file paths.*not accepted"):
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ "files": "${steps.fetch.output_files}",
+ }, ctx)
+
+
+def test_deploy_step_webdav_mode_requires_all_three_fields():
+ ctx = StepContext()
+ step = DeployStep()
+ with pytest.raises(StepValidationError, match="requires 'webdav_url', 'remote', and 'path' together"):
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ "webdav_url": "https://cloud.example.com/webdav",
+ # 'remote' and 'path' deliberately missing
+ }, ctx)
+
+
+def test_deploy_step_webdav_mode_uploads_then_deploys(monkeypatch):
+ captured = {}
+
+ def fake_upload(distributions, remote, path, webdav_url):
+ captured["upload_args"] = (distributions, remote, path, webdav_url)
+ return [{"url": "https://cloud.example.com/webdav/data/a.ttl",
+ "checksum": "abc123", "size": 100}]
+
+ def fake_deploy_from_metadata(metadata, version_id, title, abstract, description, license_url, apikey):
+ captured["deploy_metadata"] = metadata
+ captured["api_key"] = apikey
+
+ monkeypatch.setattr("databusclient.workflow.steps.webdav.upload_to_webdav", fake_upload)
+ monkeypatch.setattr("databusclient.workflow.steps.deploy_from_metadata", fake_deploy_from_metadata)
+
+ ctx = StepContext()
+ ctx.set_output("fetch", "output_files", ["/local/path/a.ttl"])
+
+ step = DeployStep()
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ "webdav_url": "https://cloud.example.com/webdav",
+ "remote": "nextcloud", "path": "datasets/mydata",
+ "files": "${steps.fetch.output_files}",
+ }, ctx)
+
+ assert captured["upload_args"][0] == ["/local/path/a.ttl"]
+ assert captured["upload_args"][1:] == ("nextcloud", "datasets/mydata", "https://cloud.example.com/webdav")
+ assert captured["api_key"] == "key123"
+ assert ctx.get_output("publish", "output_files") == ["https://cloud.example.com/webdav/data/a.ttl"]
+
+
+def test_deploy_step_classic_mode_still_works_with_urls(monkeypatch):
+ """Confirms classic mode behavior is unchanged for normal URL-based deploys."""
+ captured = {}
+
+ def fake_create_dataset(**kwargs):
+ captured["kwargs"] = kwargs
+ return {"@graph": [{"@id": "fake"}]}
+
+ def fake_deploy(dataid, api_key):
+ captured["api_key"] = api_key
+
+ monkeypatch.setattr("databusclient.workflow.steps.create_dataset", fake_create_dataset)
+ monkeypatch.setattr("databusclient.workflow.steps.api_deploy_call", fake_deploy)
+
+ ctx = StepContext()
+ step = DeployStep()
+ step.run({
+ "name": "publish", "command": "deploy",
+ "version_id": "https://databus.dbpedia.org/a/b/c/1.0",
+ "title": "T", "abstract": "A", "description": "D",
+ "license": "https://license.example.org", "api_key": "key123",
+ "files": ["https://example.org/data.ttl"],
+ }, ctx)
+
+ assert captured["kwargs"]["distributions"] == ["https://example.org/data.ttl"]
+ assert ctx.get_output("publish", "output_files") == ["https://example.org/data.ttl"]
\ No newline at end of file