Multiple Artifacts

"""
03_multi_artifact.py: Multiple artifacts produced in a single flow.

Demonstrates:
- Two separate tasks each producing a LocalArtifact
- Mixing artifact-producing tasks with normal data-passing tasks
- Independent branches that both persist output
"""

import json

from dagy import LocalArtifact, flow, task


@task
def fetch_data(source: str) -> dict:
    return {"source": source, "records": list(range(10)), "version": "v2"}


@task
def save_raw(data: dict) -> LocalArtifact:
    """Persist the raw payload as JSON."""
    return LocalArtifact(
        type="json",
        value=json.dumps(data, indent=2),
        filename="raw_data.json",
    )


@task
def transform(data: dict) -> str:
    total = sum(data["records"])
    return f"Source={data['source']}  Total={total}  Version={data['version']}"


@task
def save_summary(summary: str) -> LocalArtifact:
    """Persist the human-readable summary as plain text."""
    return LocalArtifact(type="text", value=summary, filename="summary.txt")


@flow(name="multi_artifact_flow")
def multi_artifact_flow(source: str = "warehouse") -> None:
    data = fetch_data(source)
    save_raw(data)            # artifact branch 1
    summary = transform(data) # data-passing branch
    save_summary(summary)     # artifact branch 2


if __name__ == "__main__":
    result = multi_artifact_flow.run_local(source="warehouse")
    print(f"Run ID : {result.run_id}")
    print(f"Status : {result.status}")