Tasks

Tasks are Python functions that run as part of the DAG. Use them for computation, side effects, and orchestration steps that are not loading data into a source and not producing a tracked artifact. See Python Nodes for the shared model and the SQL boundary rules.

Defining a task

Place Python files under tasks/ and decorate functions with @task:

tasks/orders.py

from sqlbuild.tasks import task, TaskContext


@task
def export_orders(ctx: TaskContext):
    rows = fetch_orders()
    return ctx.result(payload={"rows": len(rows)}, metadata={"rows": len(rows)})

The task receives a TaskContext and returns through ctx.result(...). A plain return value or None is also accepted and normalized to a successful result.

Dependencies

Tasks declare dependencies with depends_on, accepting a single function, a tuple, or a list:

@task
def fetch_orders(ctx: TaskContext):
    return ctx.result(payload=download_orders())


@task(depends_on=fetch_orders)
def summarize_orders(ctx: TaskContext):
    result = ctx.result_of(fetch_orders)
    return ctx.result(metadata={"count": len(result.payload)})

ctx.result_of(node_fn) reads the latest persisted result of an upstream node, returning a NodeResultEnvelope with payload, metadata, status, and ts fields. Results persist across runs. Use ctx.results_of(node_fn, limit=N) to read result history. Reading a missing or unsuccessful upstream raises unless you pass default=.

Tasks may depend on other tasks, assets, and loaders. They may not depend on SQL models or sources as graph dependencies, but they can read them at runtime with typed references - see SQL references.

Returning results

@task
def build_export(ctx: TaskContext):
    return ctx.result(
        payload={"path": "/exports/orders.csv"},
        metadata={"rows": 1200},
    )

• payload is the value downstream nodes read with ctx.result_of(...).
• metadata is structured JSON for catalogs and downstream reads.
• Tasks cannot set materialized - that is for assets.

Skipping

Return ctx.skip(...) to skip a task:

@task
def export_if_present(ctx: TaskContext):
    if not new_files_available():
        return ctx.skip("no new files")
    return ctx.result(payload=do_export())

• "soft" (default) skips only this task; dependents may still run if another upstream succeeded.
• "hard" skips this task and blocks its dependents.

mode accepts either a plain string or the SkipMode enum:

from sqlbuild.tasks import task, SkipMode


@task
def optional_step(ctx):
    return ctx.skip("nothing to do", mode=SkipMode.HARD)  # or mode="hard"

Retries

@task accepts a retry policy for transient failures:

from sqlbuild.retries import RetryPolicy
from sqlbuild.tasks import task


@task(retry=RetryPolicy(max_attempts=3, retry_on=(ConnectionError,)))
def call_api(ctx):
    return ctx.result(payload=fetch_from_flaky_api())

Field	Default	Description
max_attempts	3	Total attempts including the first
retry_on	Exception	Exception class, tuple, or list to retry on
initial_delay_seconds	1.0	Delay before the first retry
backoff_multiplier	2.0	Exponential backoff multiplier
max_delay_seconds	30.0	Cap on any single delay
max_elapsed_seconds	None	Overall time bound for retries
jitter	True	Randomize delays to avoid synchronized retries

The default is no retry. Set retry_on explicitly rather than relying on the broad default when you can. The original exception is preserved if all attempts fail.

Decorator parameters

Parameter	Description
name	Override the node name (defaults to the function name)
depends_on	Upstream nodes (function, tuple, or list); model()/source() for read-only SQL refs
tags	Labels for selection and grouping
group	Display/catalog grouping
description	Docs (defaults to docstring)
meta	Freeform JSON metadata
retry	A RetryPolicy

Running tasks

# Run a task and its required graph
sqb build --select export_orders --no-tests --no-audits

# Include in a full build
sqb build --select export_orders

Tasks run during sqb build. They are not validated by checks unless you also write a check that depends on them.