SDK reference

Task decorator

Both @task and @task(...) are supported.

from dagy import task


@task(
    retries=3,
    retry_delay_seconds=[1, 2, 4],
    retry_jitter_factor=0.2,
    timeout_seconds=30,
    retry_condition_fn=lambda exc: isinstance(exc, ConnectionError),
)
def fetch_record(record_id: str) -> dict:
    return {"record_id": record_id}

The example configures four total attempts. A delay list shorter than the retry count repeats its last value. Supply a nonempty list when retries are enabled. Callable delay functions are supported in local execution, but cannot be reconstructed from a serialized deployment graph; use numbers or explicit lists for portable deployments.

Local task timeouts detect a missed deadline; they do not forcibly terminate a Python thread. Running code can continue and produce side effects. Give your HTTP/database client its own timeout and make retried writes idempotent. Eager task calls do not enforce timeout_seconds.

Task methods

Mapped results preserve input order. .expand() stops at the shortest iterable; it is not a Cartesian product. With continue_on_error=True, DAG execution attempts remaining items but still fails the mapped node overall if an item fails. Eager map/expand calls use ordinary list-comprehension behavior and stop on exceptions. Local mapped items execute sequentially even with max_workers > 1.

Flow decorator

Both @flow and @flow(...) are supported. A flow accepts name, description, retries, retry_delay_seconds, retry_jitter_factor, validate_parameters, all four lifecycle-hook options, and retry_condition_fn with the same defaults as a task. It also accepts schedule=None and timezone=None.

A flow retry restarts the whole flow; a task retry repeats one task. schedule may be a cron string or ScheduleConfig. A schedule declaration takes effect only when registered for hosted execution; it does not create a local background timer.

Flow methods

run_local result fields are run_id, flow_id, flow_name, flow_version, parameters, start_time, end_time, and status. There is no result or outputs field. Read task artifacts or use an eager call when you need the Python return value.

fail_fast=False continues independent work after a task failure; work requiring failed outputs is skipped. A successfully completed call can therefore return status="FAILED". With the default fail-fast behavior, execution failure raises TaskFailedError.

Additional local-run options forwarded to the runner are cleanup_keep_last=20, skip_task_ids=None, and retry_of_run_id=None. To reuse prior results, provide both the task IDs and the earlier run ID. Reuse is possible only for available plain JSON task outputs; missing or unsupported artifacts are recomputed. Mapped and nested-flow nodes are recomputed. Retention can remove artifacts needed for reuse.

The local runner always checks flow parameters, even if validate_parameters=False is set on the decorator. It does not independently revalidate resolved task arguments. Postponed annotations are not resolved into runtime types. Treat annotations as a convenience check and validate untrusted inputs explicitly.

Deployment options

Hosted deployment results expose flow_name, flow_version, deployment_name, and skipped. Unchanged source and graph can skip registration, including when only deployment options changed; use force=True or the deployment-settings API for those changes. Schedule registration after upload is best effort: check schedules after deployment even when the artifact upload succeeded.

ScheduleConfig

from dagy import ScheduleConfig

schedule = ScheduleConfig(
    mode="cron",
    cron="0 9 * * 1-5",
    timezone="America/New_York",
    catchup="none",
)

Invalid mode, timezone, catchup, or missing mode-specific values raise ValueError. Full cron validation is available when croniter is installed; otherwise the SDK checks field count and the API performs final validation. See scheduling for execution and catchup behavior.

Hooks and state

A hook has signature hook(context: RunContext, state: State) -> None.

from dagy import RunContext, State, task


def report_attempt(context: RunContext, state: State) -> None:
    print(f"{context.name}: attempt={context.attempt}, state={state.type}")


@task(on_running=[report_attempt], on_failure=[report_attempt])
def process(value: str) -> str:
    return value.upper()

RunContext contains kind, name, one-based attempt, max_retries, parameters, and optional run_id. Do not require run_id in a task hook: it is not populated in every execution path. State contains type, optional message, and timestamp. Constructors running(), completed(), and failed() accept an optional message.

Hook exceptions can affect execution. Keep hooks short, handle failures in external reporting systems, and never log credentials. These are in-process hooks, not hosted webhook subscriptions.

Local artifacts

from dagy import LocalArtifact, task


@task
def export_report() -> LocalArtifact:
    return LocalArtifact(type="text", value="Paid orders: 2", filename="report.txt")

LocalArtifact(type, value, filename=None) supports json, text, and file. For file, value is the path of an existing file to copy. Plain JSON task outputs are stored without a wrapper. A LocalArtifact passed downstream remains a wrapper, so downstream code must explicitly use .value where appropriate.

The local per-artifact default is 5 MiB, configurable with DAGY_LOCAL_ARTIFACT_MAX_BYTES. Serialization or size failures log a warning and may leave a task marked successful without a persisted output. Check the artifact when you depend on it for inspection or retry reuse.

Packaging helpers

build_artifact(flow, *args, output_dir=None, flow_version="1", **kwargs) builds a deployment ZIP and returns a BuildArtifact. Read .artifact_path, .metadata_path, .spec_path, and .manifest_path on the result. The default output directory is ~/.dagy/builds; this default does not follow DAGY_LOCAL_DIR.

from dagy import build_artifact
from order_pipeline import order_pipeline

artifact = build_artifact(order_pipeline, output_dir="dist", source="sample")
print(artifact.artifact_path)

The package includes the flow source file and graph metadata. Dependency declarations are recorded, not vendored or installed. Imported project modules and data assets are not collected automatically.

For a separately built artifact, deploy_artifact accepts explicit upload details:

import os
from dagy import build_artifact, deploy_artifact
from order_pipeline import order_pipeline

artifact = build_artifact(order_pipeline, output_dir="dist")
result = deploy_artifact(
    artifact_path=str(artifact.artifact_path),
    deployment_name="orders-develop",
    flow_name="order_pipeline",
    flow_version="1",
    api_url=os.environ["DAGY_API_URL"],
    environment="develop",
)
print(result.deployment_name)

Authenticate first through the CLI login flow. Additional options are access_token=None, status="ACTIVE", schedule=None, timezone=None, default_executor=None, tags=None, namespace=None, dep_package_slugs=None, and execution_mode="nano". The helper returns a DeployResult with deployment and flow identifiers, and additional metadata returned by the API. Deprecated bucket and prefix arguments are ignored. Use the deployment guide for version selection and the recommended end-to-end process.

DagyClient

DagyClient(base_url, token=None, timeout=10, user_agent=None) is a synchronous JSON HTTP client. token accepts a session token or an API key and sends it as Authorization: Bearer ...; without one the client reads saved CLI credentials. An organization header is read from saved CLI credentials, while an API key remains bound to its creation workspace. There is no separate api_key constructor argument and no automatic token refresh. See API-key authentication for required scopes. For automation, pass token=os.environ["DAGY_SERVICE_KEY"] explicitly after importing os.

The client does not automatically paginate or retry general API requests. Upload helpers handle multipart retries separately. Use listing flows for a pagination example and the HTTP reference for request and response definitions.

SDK errors

See quality and contract examples, API errors, and troubleshooting.