Documentation Index
Fetch the complete documentation index at: https://docs.sqlbuild.com/llms.txt
Use this file to discover all available pages before exploring further.
SqlBuildProject
Project metadata and DAG artifact preparation.
from sqlbuild.integrations.dagster import SqlBuildProject
project = SqlBuildProject(
project_dir=".",
target_path="target",
dag_filename="sqlbuild_dag.json",
sqb_command=("sqb",),
prepare_project_cli_args=("compile", "--dag"),
)
| Parameter | Default | Description |
|---|
project_dir | (required) | Path to the SQLBuild project root |
target_path | "target" | Relative path to the target directory |
dag_filename | "sqlbuild_dag.json" | DAG artifact filename |
sqb_command | ("sqb",) | Command to invoke SQLBuild CLI |
prepare_project_cli_args | ("compile", "--dag") | CLI args used by prepare() to generate the DAG |
Methods
| Method | Description |
|---|
dag_path | Property. Returns the full path to the DAG artifact (project_dir / target_path / dag_filename). |
prepare() | Runs sqb compile --dag <dag_path> to generate the DAG artifact. Raises DagsterProjectPrepareError on failure. |
prepare_if_dev() | Calls prepare() only when the DAGSTER_IS_DEV_CLI environment variable is set. |
SqlBuildCliResource
Dagster ConfigurableResource that shells out to the SQLBuild CLI.
from sqlbuild.integrations.dagster import SqlBuildCliResource, SqlBuildProject
# From a project
resource = SqlBuildCliResource(SqlBuildProject(project_dir="."))
# Or with explicit paths
resource = SqlBuildCliResource(
project_dir=".",
sqb_command=["sqb"],
dag_path="target/sqlbuild_dag.json",
)
| Parameter | Default | Description |
|---|
project_dir | "." | Path to the SQLBuild project root, or a SqlBuildProject instance |
sqb_command | ["sqb"] | Command to invoke SQLBuild CLI |
dag_path | None | Path to the DAG artifact. When set, enables asset selection bridging and structured event emission. |
cli()
Create a CLI invocation for the provided command arguments.
invocation = resource.cli(
["build"],
context=context, # Dagster execution context (optional)
raise_on_error=True, # raise Failure on non-zero exit (default)
)
SqlBuildCliInvocation.
SqlBuildCliInvocation
A running or completed SQLBuild CLI subprocess.
Methods
| Method | Description |
|---|
wait() | Wait for the process to complete. Streams stdout/stderr in real time. Returns self. |
stream() | Wait for the process, then yield Dagster MaterializeResult and AssetCheckResult events parsed from execution JSON. |
is_successful() | Returns True if exit code is 0. |
get_error() | Returns a Dagster Failure if the process failed, None otherwise. |
get_artifact(name) | Read a JSON artifact from the project’s target/ directory. |
Properties
| Property | Description |
|---|
stdout | Captured stdout after wait() or stream(). |
stderr | Captured stderr after wait() or stream(). |
returncode | Process exit code after completion. |
execution_payload | Parsed execution JSON, if available. |
stream() behavior
stream() is the primary way to emit Dagster events from a SQLBuild execution:
- Waits for the subprocess to complete while streaming output to stdout/stderr
- Reads the structured execution JSON (from
--json-output tempfile)
- Maps each completed asset to a
MaterializeResult with metadata (status, duration, row counts)
- Maps each check result to an
AssetCheckResult with pass/fail, severity, and check metadata
- If the DAG artifact is available, results are matched to the correct Dagster asset keys
- Raises
Failure if the process exited non-zero and raise_on_error is set
sqlbuild_assets
Decorator that creates a Dagster multi-asset definition from a SQLBuild DAG artifact.
from sqlbuild.integrations.dagster import sqlbuild_assets
@sqlbuild_assets(
project=project, # or dag=path_or_dict
translator=MyTranslator(), # optional
name="my_sqlbuild_assets", # optional
include_scenario_checks=True, # default: True
required_resource_keys={"sqb"}, # optional
)
def my_assets(context, sqb):
yield from sqb.cli(["build"], context=context).stream()
| Parameter | Default | Description |
|---|
project | None | SqlBuildProject instance. Mutually exclusive with dag. |
dag | None | DAG artifact as a path, string, or pre-loaded dict. Mutually exclusive with project. |
translator | SqlBuildDagsterTranslator() | Translator for asset keys, groups, tags, and metadata. |
name | None | Override the Dagster multi-asset name. |
include_scenario_checks | True | Whether to include scenario checks alongside test and audit checks. |
required_resource_keys | None | Additional Dagster resource keys to require. |
project nor dag is provided, the decorator looks for target/sqlbuild_dag.json in the current directory.
sqlbuild_scenario_checks
Decorator that creates a Dagster multi-asset-check definition for SQLBuild scenarios only. Useful when you want to run scenarios separately from the main build.
from sqlbuild.integrations.dagster import sqlbuild_scenario_checks
@sqlbuild_scenario_checks(project=project)
def my_scenario_checks(context, sqb):
yield from sqb.cli(["scenario", "test"], context=context).stream()
sqlbuild_assets except there is no include_scenario_checks flag.
SqlBuildDagsterTranslator
Customise how SQLBuild DAG nodes map to Dagster asset metadata. Subclass and override any method.
Methods
| Method | Arguments | Returns | Default behavior |
|---|
get_asset_key | node | AssetKey | AssetKey(node["asset_key"]) |
get_group_name | node | str | None | Node kind (e.g. "model", "source") |
get_tags | node | dict[str, str] | sqlbuild/kind tag plus any model tags |
get_metadata | node | dict[str, Any] | sqlbuild_id, sqlbuild_name, sqlbuild_kind, path, target, description, columns |
get_description | node | str | None | Node description if present |
get_check_name | check | str | "{kind}__{name}" with optional column/target suffix |
get_check_metadata | check | dict[str, Any] | Check ID, kind, name, and selector |
Node structure
Each node dict passed to translator methods contains:
{
"id": "model:fact_orders",
"kind": "model",
"name": "fact_orders",
"asset_key": ["dev", "fact_orders"],
"target": {
"database": null,
"schema": "dev",
"name": "fact_orders",
"qualified_name": "dev.fact_orders"
},
"path": "models/marts/fact_orders.sql",
"description": "Order fact table with waffle and payment details.",
"tags": ["marts"],
"columns": [
{"name": "order_id", "type": "INTEGER"}
],
"materialization_type": "table"
}
Check structure
Each check dict passed to check translator methods contains:
{
"id": "audit:not_null:model:fact_orders:order_id",
"kind": "audit",
"name": "not_null",
"checked_asset_ids": ["model:fact_orders"],
"path": "audits/generic/not_null.sql",
"severity": "error",
"attached_target_name": "fact_orders",
"attached_column_name": "order_id"
}
{
"id": "scenario:daily_revenue_minimal",
"kind": "scenario",
"name": "daily_revenue_minimal",
"checked_asset_ids": ["model:daily_revenue"],
"path": "tests/scenarios/revenue/daily_revenue_minimal.sql",
"assertion_names": ["no_negative_revenue"],
"expected_model_names": ["daily_revenue"],
"fixture_refs": ["stg_orders", "stg_payments"]
}
