Build & Deploy

"""
01_build_and_deploy.py: Build an artifact and deploy it to DAGY Cloud.

Demonstrates:
- build_artifact(flow): package the flow into a deployable artifact
- deploy_artifact(...): register the artifact with the DAGY API
- DagyClient.trigger_run(...): kick off a remote run

Prerequisites:
  export DAGY_API_URL=https://api.dagy.io
  export DAGY_TOKEN=<your_access_token>

Running without DAGY_API_URL set will skip the deploy/trigger steps and
only show the local artifact build.
"""

import os

from dagy import DagyClient, LocalArtifact, build_artifact, deploy_artifact, flow, task
from dagy.deploy import resolve_artifact_bucket_name


@task
def fetch(source: str) -> dict:
    return {"source": source, "value": 42}


@task
def process(payload: dict) -> LocalArtifact:
    import json
    return LocalArtifact(
        type="json",
        value=json.dumps(payload),
        filename="output.json",
    )


@flow(name="deployable_flow", description="Example deployable flow")
def deployable_flow(source: str = "cloud") -> None:
    data = fetch(source)
    process(data)


if __name__ == "__main__":
    # -- Step 1: Build --------------------------------------------------------
    artifact = build_artifact(deployable_flow, source="cloud")
    print(f"[build]  artifact_path={artifact.artifact_path}")

    api_url = os.getenv("DAGY_API_URL")
    access_token = os.getenv("DAGY_TOKEN")

    if not api_url:
        print("[deploy] DAGY_API_URL not set - skipping deploy and trigger.")
        print("[local]  Running locally instead...")
        result = deployable_flow.run_local(source="local")
        print(f"[local]  run_id={result.run_id}  status={result.status}")
    else:
        # -- Step 2: Deploy ---------------------------------------------------
        deployment_name = os.getenv("DAGY_DEPLOYMENT_NAME", "deployable-flow-prod")
        deployed = deploy_artifact(
            artifact_path=str(artifact.artifact_path),
            deployment_name=deployment_name,
            flow_name=deployable_flow.name,
            flow_version=deployable_flow.version,
            bucket=resolve_artifact_bucket_name(),
            api_url=api_url,
            access_token=access_token,
            schedule=os.getenv("DAGY_SCHEDULE"),   # e.g. "0 6 * * *" for 6am daily
            status="ACTIVE",
        )
        print(
            f"[deploy] flow={deployed.flow_name}:{deployed.flow_version}"
            f"  s3={deployed.artifact_s3_uri}"
        )

        # -- Step 3: Trigger a remote run -------------------------------------
        client = DagyClient(api_url, token=access_token)
        run = client.trigger_run({
            "flow_name": deployable_flow.name,
            "flow_version": deployable_flow.version,
            "parameters": {"source": "cloud"},
        })
        print(f"[trigger] remote run started: {run}")