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.
SQLBuild includes a Rivers integration that maps your project’s models, sources, seeds, loaders, and functions into Rivers assets with dependency edges preserved. SQLBuild handles the SQL transformation layer. Rivers handles scheduling, execution, and the asset-centric UI.
Install
pip install 'sqlbuild[rivers]'
# or
uv pip install 'sqlbuild[rivers]'
rivers alongside SQLBuild.
Try it
sqb playground --template rivers
cd sqlbuild-playground
uv run rivers dev rivers_pipeline.definitions
rivers_pipeline/definitions.py that includes asset definitions and a configured job. Open the Rivers UI to inspect assets and trigger materializations.
How it works
sqb compile --dag generates a static sqlbuild_dag.json artifact with your project’s full graph (nodes, edges)@sqlbuild_assets() reads the artifact and creates one Rivers AssetDef per source, loader, seed, model, and function, with dependency edges preserved- The decorated function runs
sqb build as a subprocess and yields Materialization events for each output
Quickstart
# definitions.py
from __future__ import annotations
import subprocess
from collections.abc import Iterator
from pathlib import Path
from typing import Any
import rivers as rs
from sqlbuild.integrations.rivers import SqlBuildProject, sqlbuild_assets
PROJECT_DIR = Path(__file__).resolve().parent.parent
SQLBUILD_PROJECT = SqlBuildProject(project_dir=PROJECT_DIR)
SQLBUILD_PROJECT.prepare_if_dev()
@sqlbuild_assets(project=SQLBUILD_PROJECT)
def my_assets(context: Any) -> Iterator[Any]:
completed = subprocess.run(
["sqb", "build"],
cwd=PROJECT_DIR,
capture_output=True,
check=False,
text=True,
)
if completed.returncode != 0:
raise RuntimeError(completed.stderr or completed.stdout)
for output_name in context.output_selection:
yield rs.Materialization(output_name=output_name)
repo = rs.CodeRepository(
assets=[my_assets],
jobs=[
rs.Job(
name="sqlbuild_pipeline",
assets=[my_assets],
executor=rs.Executor.in_process(),
),
],
)
SqlBuildProject
SqlBuildProject manages paths and DAG artifact generation:
project = SqlBuildProject(project_dir=Path("."))
| Field | Default | Description |
|---|
project_dir | required | Path to the SQLBuild project root |
target_path | target | Directory for build artifacts |
dag_filename | sqlbuild_dag.json | DAG artifact filename |
sqb_command | ("sqb",) | Command to invoke SQLBuild |
prepare_project_cli_args | ("compile", "--dag") | CLI arguments for DAG generation |
Methods
| Method | Description |
|---|
prepare() | Generate the DAG artifact by running sqb compile --dag |
prepare_if_dev() | Generate the DAG artifact only when RIVERS_DEPLOYMENT=dev |
dag_path | Property returning the full path to the DAG artifact |
prepare_if_dev() so the DAG artifact is auto-generated during local development but not during production deployments where it should already exist.
sqlbuild_assets
The @sqlbuild_assets() decorator creates a Rivers multi-asset definition from a SQLBuild DAG artifact:
@sqlbuild_assets(project=project)
def my_assets(context):
...
| Parameter | Description |
|---|
project | SqlBuildProject instance (generates and locates the DAG artifact) |
dag | Alternative: pass a DAG path or dict directly instead of a project |
translator | Optional SqlBuildRiversTranslator for custom name/tag/metadata mapping |
name | Optional asset name override |
project or dag, not both.
SqlBuildRiversTranslator
Override the default mapping from SQLBuild DAG nodes to Rivers asset metadata:
from sqlbuild.integrations.rivers import SqlBuildRiversTranslator
class CustomTranslator(SqlBuildRiversTranslator):
def get_group_name(self, node):
return "my_custom_group"
def get_tags(self, node):
tags = super().get_tags(node)
tags.append("team:analytics")
return tags
@sqlbuild_assets(project=project, translator=CustomTranslator())
def my_assets(context):
...
Translator methods
| Method | Default behavior |
|---|
get_asset_name(node) | Joins the asset key parts with __ |
get_group_name(node) | Uses the project name |
get_tags(node) | Includes sqlbuild/kind:<kind> plus model tags |
get_kinds(node) | Returns ["sqlbuild", "<materialization_type>"] |
get_metadata(node) | Includes SQLBuild name, kind, path, target, columns, etc. |
Running locally
Run the pipeline directly without the Rivers UI:
uv run python rivers_pipeline/definitions.py
uv run rivers dev rivers_pipeline.definitions
