This library provides a Dagster integration with dbt (data build tool), created by dbt Labs.
Loads a set of dbt models from a dbt project into Dagster assets.
Creates one Dagster asset for each dbt model. All assets will be re-materialized using a single dbt run or dbt build command.
project_dir (Optional[str]) – The directory containing the dbt project to load.
profiles_dir (Optional[str]) – The profiles directory to use for loading the DBT project. Defaults to a directory called “config” inside the project_dir.
target_dir (Optional[str]) – The target directory where dbt will place compiled artifacts. Defaults to “target” underneath the project_dir.
select (Optional[str]) – A dbt selection string for the models in a project that you want to include. Defaults to “*”.
exclude (Optional[str]) – A dbt selection string for the models in a project that you want to exclude. Defaults to “”.
key_prefix (Optional[Union[str, List[str]]]) – A prefix to apply to all models in the dbt project. Does not apply to sources.
source_key_prefix (Optional[Union[str, List[str]]]) – A prefix to apply to all sources in the dbt project. Does not apply to models.
runtime_metadata_fn – (Optional[Callable[[SolidExecutionContext, Mapping[str, Any]], Mapping[str, Any]]]): A function that will be run after any of the assets are materialized and returns metadata entries for the asset, to be displayed in the asset catalog for that run.
io_manager_key (Optional[str]) – The IO manager key that will be set on each of the returned assets. When other ops are downstream of the loaded assets, the IOManager specified here determines how the inputs to those ops are loaded. Defaults to “io_manager”.
node_info_to_asset_key – (Mapping[str, Any] -> AssetKey): A function that takes a dictionary of dbt metadata and returns the AssetKey that you want to represent a given model or source. By default: dbt model -> AssetKey([model_name]) and dbt source -> AssetKey([source_name, table_name])
use_build_command (bool) – Flag indicating if you want to use dbt build as the core computation for this asset, rather than dbt run.
partitions_def (Optional[PartitionsDefinition]) – Defines the set of partition keys that compose the dbt assets.
partition_key_to_vars_fn (Optional[str -> Dict[str, Any]]) – A function to translate a given partition key (e.g. ‘2022-01-01’) to a dictionary of vars to be passed into the dbt invocation (e.g. {“run_date”: “2022-01-01”})
node_info_to_group_fn (Dict[str, Any] -> Optional[str]) – A function that takes a dictionary of dbt node info and returns the group that this node should be assigned to.
display_raw_sql (Optional[bool]) – [Experimental] A flag to indicate if the raw sql associated with each model should be included in the asset description. For large projects, setting this flag to False is advised to reduce the size of the resulting snapshot.
Loads a set of dbt models, described in a manifest.json, into Dagster assets.
Creates one Dagster asset for each dbt model. All assets will be re-materialized using a single dbt run command.
manifest_json (Optional[Mapping[str, Any]]) – The contents of a DBT manifest.json, which contains a set of models to load into assets.
select (Optional[str]) – A dbt selection string for the models in a project that you want to include. Defaults to “*”.
exclude (Optional[str]) – A dbt selection string for the models in a project that you want to exclude. Defaults to “”.
key_prefix (Optional[Union[str, List[str]]]) – A prefix to apply to all models in the dbt project. Does not apply to sources.
source_key_prefix (Optional[Union[str, List[str]]]) – A prefix to apply to all sources in the dbt project. Does not apply to models.
runtime_metadata_fn – (Optional[Callable[[SolidExecutionContext, Mapping[str, Any]], Mapping[str, Any]]]): A function that will be run after any of the assets are materialized and returns metadata entries for the asset, to be displayed in the asset catalog for that run.
io_manager_key (Optional[str]) – The IO manager key that will be set on each of the returned assets. When other ops are downstream of the loaded assets, the IOManager specified here determines how the inputs to those ops are loaded. Defaults to “io_manager”.
selected_unique_ids (Optional[Set[str]]) – The set of dbt unique_ids that you want to load as assets.
node_info_to_asset_key – (Mapping[str, Any] -> AssetKey): A function that takes a dictionary of dbt node info and returns the AssetKey that you want to represent that node. By default, the asset key will simply be the name of the dbt model.
use_build_command (bool) – Flag indicating if you want to use dbt build as the core computation for this asset, rather than dbt run.
partitions_def (Optional[PartitionsDefinition]) – Defines the set of partition keys that compose the dbt assets.
partition_key_to_vars_fn (Optional[str -> Dict[str, Any]]) – A function to translate a given partition key (e.g. ‘2022-01-01’) to a dictionary of vars to be passed into the dbt invocation (e.g. {“run_date”: “2022-01-01”})
node_info_to_group_fn (Dict[str, Any] -> Optional[str]) – A function that takes a dictionary of dbt node info and returns the group that this node should be assigned to.
display_raw_sql (Optional[bool]) – [Experimental] A flag to indicate if the raw sql associated with each model should be included in the asset description. For large projects, setting this flag to False is advised to reduce the size of the resulting snapshot.
A resource that allows you to execute dbt cli commands. For the most up-to-date documentation on the specific parameters available to you for each command, check out the dbt docs:
https://docs.getdbt.com/reference/commands/run
To use this as a dagster resource, we recommend using
dbt_cli_resource.
Run the build command on a dbt project. kwargs are passed in as additional parameters.
select (List[str], optional) – the models/resources to include in the run.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
default flags that were configured on resource initialization (if any) overriding the default values if necessary.
command (str) – The command you wish to run (e.g. ‘run’, ‘test’, ‘docs generate’, etc.)
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the compile command on a dbt project. kwargs are passed in as additional parameters.
models (List[str], optional) – the models to include in compilation.
exclude (List[str]), optional) – the models to exclude from compilation.
select (List[str], optional) – the models to include in compilation.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the source snapshot-freshness command on a dbt project. kwargs are passed in as additional parameters.
select (List[str], optional) – the sources to include in the run.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the docs generate command on a dbt project. kwargs are passed in as additional parameters.
compile_project (bool, optional) – If true, compile the project before generating a catalog.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Get a parsed version of the manifest.json file for the relevant dbt project.
for this dbt project.
Dict[str, Any]
Get a parsed version of the run_results.json file for the relevant dbt project.
for this dbt project.
Dict[str, Any]
Run the ls command on a dbt project. kwargs are passed in as additional parameters.
select (List[str], optional) – the resources to include in the output.
models (List[str], optional) – the models to include in the output.
exclude (List[str], optional) – the resources to exclude from the output.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Remove the run_results.json file from previous runs (if it exists).
Run the run command on a dbt project. kwargs are passed in as additional parameters.
models (List[str], optional) – the models to include in the run.
exclude (List[str]), optional) – the models to exclude from the run.
select (List[str], optional) – the models to include in the run.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the run-operation command on a dbt project. kwargs are passed in as additional parameters.
macro (str) – the dbt macro to invoke.
args (Dict[str, Any], optional) – the keyword arguments to be supplied to the macro.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the seed command on a dbt project. kwargs are passed in as additional parameters.
show (bool, optional) – If True, then show a sample of the seeded data in the
response. Defaults to False.
select (List[str], optional) – the snapshots to include in the run.
exclude (List[str], optional) – the snapshots to exclude from the run.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the snapshot command on a dbt project. kwargs are passed in as additional parameters.
select (List[str], optional) – the snapshots to include in the run.
exclude (List[str], optional) – the snapshots to exclude from the run.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
Run the test command on a dbt project. kwargs are passed in as additional parameters.
models (List[str], optional) – the models to include in testing.
exclude (List[str], optional) – the models to exclude from testing.
data (bool, optional) – If True (default), then run data tests.
schema (bool, optional) – If True (default), then run schema tests.
select (List[str], optional) – the models to include in testing.
DbtCliOutput containingparsed log output as well as the contents of run_results.json (if applicable).
The results of executing a dbt command, along with additional metadata about the dbt CLI process that was run.
Note that users should not construct instances of this class directly. This class is intended to be constructed from the JSON output of dbt commands.
The full shell command that was executed.
str
The return code of the dbt CLI process.
int
The raw output (stdout) of the dbt CLI process.
str
List of parsed JSON logs produced by the dbt command.
List[Dict[str, Any]]
Dictionary containing dbt-reported result information contained in run_results.json. Some dbt commands do not produce results, and will therefore have result = None.
Optional[Dict[str, Any]]
Hostname where dbt docs are being served for this project.
Optional[str]
Which directory to look in for the dbt_project.yml file. Default is the current working directory and its parents.
Default Value: ‘.’
Which directory to look in for the profiles.yml file. Default = $DBT_PROFILES_DIR or $HOME/.dbt
Which profile to load. Overrides setting in dbt_project.yml.
Which target to load for the given profile.
Supply variables to the project. This argument overrides variables defined in your dbt_project.yml file. This argument should be a dictionary, eg. {‘my_variable’: ‘my_value’}
If set, bypass the adapter-level cache of database state
Default Value: False
If dbt would normally warn, instead raise an exception. Examples include –models that selects nothing, deprecations, configurations with no associated models, invalid test configurations, and missing sources/refs in tests.
Default Value: False
Path to the dbt executable. Default is dbt
Default Value: ‘dbt’
When True, will not raise an exception when the dbt CLI returns error code 1. Default is False.
Default Value: False
The directory path for target if different from the default target-path in your dbt project configuration file.
Default Value: ‘target’
The url for where dbt docs are being served for this project.
When True, dbt will invoked with the –log-format json flag, allowing Dagster to parse the log messages and emit simpler log messages to the event log.
Default Value: True
When True, logs emitted from dbt will be logged to the Dagster event log.
Default Value: True
This resource issues dbt CLI commands against a configured dbt project.
This class exposes methods on top of the dbt Cloud REST API v2.
For a complete set of documentation on the dbt Cloud Administrative REST API, including expected response JSON schemae, see the dbt Cloud API Docs.
dbt Cloud API Token. User tokens can be found in the [dbt Cloud UI](https://cloud.getdbt.com/#/profile/api/), or see the [dbt Cloud Docs](https://docs.getdbt.com/docs/dbt-cloud/dbt-cloud-api/service-tokens) for instructions on creating a Service Account token.
dbt Cloud Account ID. This value can be found in the url of a variety of views in the dbt Cloud UI, e.g. https://cloud.getdbt.com/#/accounts/{account_id}/settings/.
Specifies if you would like any job that is triggered using this resource to automatically disable its schedule.
Default Value: True
The maximum number of times requests to the dbt Cloud API should be retried before failing.
Default Value: 3
Time (in seconds) to wait between each request retry.
Default Value: 0.25
The hostname where dbt cloud is being hosted (e.g. https://my_org.cloud.getdbt.com/).
Default Value: ‘https://cloud.getdbt.com/’
This resource allows users to programatically interface with the dbt Cloud Administrative REST API (v2) to launch jobs and monitor their progress. This currently implements only a subset of the functionality exposed by the API.
For a complete set of documentation on the dbt Cloud Administrative REST API, including expected response JSON schemae, see the dbt Cloud API Docs.
To configure this resource, we recommend using the configured method.
Examples:
from dagster import job
from dagster_dbt import dbt_cloud_resource
my_dbt_cloud_resource = dbt_cloud_resource.configured(
{
"auth_token": {"env": "DBT_CLOUD_AUTH_TOKEN"},
"account_id": {"env": "DBT_CLOUD_ACCOUNT_ID"},
}
)
@job(resource_defs={"dbt_cloud": my_dbt_cloud_resource})
def my_dbt_cloud_job():
...
A client for a dbt RPC server.
To use this as a dagster resource, we recommend using
dbt_rpc_resource.
The output from executing a dbt command via the dbt RPC server.
The parsed contents of the “result” field of the JSON response from the rpc server (if any).
Dict[str, Any]
The entire contents of the JSON response from the rpc server.
Dict[str, Any]
The original Response from which this output was generated.
requests.Response
This resource defines a dbt RPC client for an RPC server running on 0.0.0.0:8580.
Default Value: 8580
This resource defines a dbt RPC client.
To configure this resource, we recommend using the configured method.
Examples:
Examples:
from dagster_dbt import dbt_rpc_resource
custom_dbt_rpc_resource = dbt_rpc_resource.configured({"host": "80.80.80.80","port": 8080,})
@job(resource_defs={"dbt_rpc": custom_dbt_rpc_sync_resource})
def dbt_rpc_job():
# Run ops with `required_resource_keys={"dbt_rpc", ...}`.
Default Value: 8580
Default Value: 1
This resource defines a synchronous dbt RPC client, which sends requests to a dbt RPC server, and waits for the request to complete before returning.
To configure this resource, we recommend using the configured method.
Examples:
from dagster_dbt import dbt_rpc_sync_resource
custom_sync_dbt_rpc_resource = dbt_rpc_sync_resource.configured({"host": "80.80.80.80","port": 8080,})
@job(resource_defs={"dbt_rpc": custom_dbt_rpc_sync_resource})
def dbt_rpc_sync_job():
# Run ops with `required_resource_keys={"dbt_rpc", ...}`.
If you’re using asset-based dbt APIs like load_assets_from_dbt_project, you usually will not also use the below op-based APIs.
dagster_dbt provides a set of pre-built ops that work with either the CLI or RPC interfaces. For
more advanced use cases, we suggest building your own ops which directly interact with these resources.
If True, materializations corresponding to the results of the dbt operation will be yielded when the op executes. Default: True
Default Value: True
If provided and yield_materializations is True, these components will be used to prefix the generated asset keys.
Default Value: [‘dbt’]
This op executes a dbt run command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_run_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_run_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_run_op()
This op executes a dbt compile command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_compile_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_compile_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_compile_op()
This op executes a dbt ls command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_ls_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_ls_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_ls_op()
This op executes a dbt test command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_test_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_test_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_test_op()
This op executes a dbt snapshot command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_snapshot_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_snapshot_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_snapshot_op()
This op executes a dbt seed command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_seed_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_seed_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_seed_op()
This op executes a dbt docs generate command. It requires the use of a dbt resource, which can be
set to execute this command through the CLI (using the dbt_cli_resource) or
over RPC (using the dbt_rpc_sync_resource).
Examples:
from dagster import job
from dagster_dbt import dbt_docs_generate_op, dbt_cli_resource, dbt_rpc_sync_resource
@job(resource_defs={"dbt":dbt_cli_resource})
def my_dbt_cli_job():
dbt_docs_generate_op()
@job(resource_defs={"dbt":dbt_rpc_sync_resource})
def my_dbt_rpc_job():
dbt_docs_generate_op()
The integer ID of the relevant dbt Cloud job. You can find this value by going to the details page of your job in the dbt Cloud UI. It will be the final number in the url, e.g.: https://cloud.getdbt.com/#/accounts/{account_id}/projects/{project_id}/jobs/{job_id}/
The time (in seconds) that will be waited between successive polls.
Default Value: 10
The maximum time that will waited before this operation is timed out. By default, this will never time out.
Default Value: None
If True, materializations corresponding to the results of the dbt operation will be yielded when the op executes.
Default Value: True
If provided and yield_materializations is True, these components will be used to prefix the generated asset keys.
Default Value: [‘dbt’]
Initiates a run for a dbt Cloud job, then polls until the run completes. If the job fails or is otherwised stopped before succeeding, a dagster.Failure exception will be raised, and this op will fail.
It requires the use of a ‘dbt_cloud’ resource, which is used to connect to the dbt Cloud API.
Config Options:
The integer ID of the relevant dbt Cloud job. You can find this value by going to the details
page of your job in the dbt Cloud UI. It will be the final number in the url, e.g.:
https://cloud.getdbt.com/#/accounts/{account_id}/projects/{project_id}/jobs/{job_id}/
The time (in seconds) that will be waited between successive polls. Defaults to 10.
The maximum time (in seconds) that will waited before this operation is timed out. By default, this will never time out.
If True, materializations corresponding to the results of the dbt operation will be
yielded when the solid executes. Defaults to True.
If provided and yield_materializations is True, these components will be used to ” prefix the generated asset keys. Defaults to [“dbt”].
Examples:
from dagster import job
from dagster_dbt import dbt_cloud_resource, dbt_cloud_run_op
my_dbt_cloud_resource = dbt_cloud_resource.configured(
{"auth_token": {"env": "DBT_CLOUD_AUTH_TOKEN"}, "account_id": 77777}
)
run_dbt_nightly_sync = dbt_cloud_run_op.configured(
{"job_id": 54321}, name="run_dbt_nightly_sync"
)
@job(resource_defs={"dbt_cloud": my_dbt_cloud_resource})
def dbt_cloud():
run_dbt_nightly_sync()
The base exception of the dagster-dbt library.
Represents an error while executing a dbt CLI command.
Represents a fatal error in the dbt CLI (return code 2).
Represents a model error reported by the dbt CLI at runtime (return code 1).
Represents a problem in finding the target/run_results.json artifact when executing a dbt
CLI command.
For more details on target/run_results.json, see
https://docs.getdbt.com/reference/dbt-artifacts#run_resultsjson.
This function yields dagster.AssetMaterialization events for each model updated by
a dbt command.
Information parsed from a DbtOutput object.
Note that this will not work with output from the dbt_rpc_resource, because this resource does not wait for a response from the RPC server before returning. Instead, use the dbt_rpc_sync_resource, which will wait for execution to complete.
Examples:
from dagster import op, Output
from dagster_dbt.utils import generate_materializations
from dagster_dbt import dbt_cli_resource, dbt_rpc_sync_resource
@op(required_resource_keys={"dbt"})
def my_custom_dbt_run(context):
dbt_output = context.resources.dbt.run()
for materialization in generate_materializations(dbt_output):
# you can modify the materialization object to add extra metadata, if desired
yield materialization
yield Output(my_dbt_output)
@job(resource_defs={{"dbt":dbt_cli_resource}})
def my_dbt_cli_job():
my_custom_dbt_run()
@job(resource_defs={{"dbt":dbt_rpc_sync_resource}})
def my_dbt_rpc_job():
my_custom_dbt_run()