dbt (dagster-dbt)¶

Assets (dbt Core)¶

dagster_dbt.load_assets_from_dbt_project(project_dir, profiles_dir=None, target_dir=None, select=None, exclude=None, key_prefix=None, source_key_prefix=None, runtime_metadata_fn=None, io_manager_key=None, node_info_to_asset_key=<function _get_node_asset_key>, use_build_command=False, partitions_def=None, partition_key_to_vars_fn=None, node_info_to_group_fn=<function _get_node_group_name>, node_info_to_freshness_policy_fn=<function _get_node_freshness_policy>, node_info_to_definition_metadata_fn=<function _get_node_metadata>, display_raw_sql=None, dbt_resource_key='dbt')[source]¶

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.

Parameters:

• 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.

• dbt_resource_key (Optional[str]) – The resource key that the dbt resource will be specified at. Defaults to “dbt”.

• 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.

• node_info_to_freshness_policy_fn (Dict[str, Any] -> Optional[FreshnessPolicy]) – A function that takes a dictionary of dbt node info and optionally returns a FreshnessPolicy that should be applied to this node. By default, freshness policies will be created from config applied to dbt models, i.e.: dagster_freshness_policy={“maximum_lag_minutes”: 60, “cron_schedule”: “0 9 * * *”} will result in that model being assigned FreshnessPolicy(maximum_lag_minutes=60, cron_schedule=”0 9 * * *”)

• node_info_to_definition_metadata_fn (Dict[str, Any] -> Optional[Dict[str, MetadataUserInput]]) – A function that takes a dictionary of dbt node info and optionally returns a dictionary of metadata to be attached to the corresponding definition. This is added to the default metadata assigned to the node, which consists of the node’s schema (if present).

• 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.

dagster_dbt.load_assets_from_dbt_manifest(manifest_json, select=None, exclude=None, key_prefix=None, source_key_prefix=None, runtime_metadata_fn=None, io_manager_key=None, selected_unique_ids=None, node_info_to_asset_key=<function _get_node_asset_key>, use_build_command=False, partitions_def=None, partition_key_to_vars_fn=None, node_info_to_group_fn=<function _get_node_group_name>, node_info_to_freshness_policy_fn=<function _get_node_freshness_policy>, node_info_to_definition_metadata_fn=<function _get_node_metadata>, display_raw_sql=None, dbt_resource_key='dbt')[source]¶

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.

Parameters:

• 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.

• dbt_resource_key (Optional[str]) – The resource key that the dbt resource will be specified at. Defaults to “dbt”.

• 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.

• node_info_to_freshness_policy_fn (Dict[str, Any] -> Optional[FreshnessPolicy]) – A function that takes a dictionary of dbt node info and optionally returns a FreshnessPolicy that should be applied to this node. By default, freshness policies will be created from config applied to dbt models, i.e.: dagster_freshness_policy={“maximum_lag_minutes”: 60, “cron_schedule”: “0 9 * * *”} will result in that model being assigned FreshnessPolicy(maximum_lag_minutes=60, cron_schedule=”0 9 * * *”)

• node_info_to_definition_metadata_fn (Dict[str, Any] -> Optional[Dict[str, MetadataUserInput]]) – A function that takes a dictionary of dbt node info and optionally returns a dictionary of metadata to be attached to the corresponding definition. This is added to the default metadata assigned to the node, which consists of the node’s schema (if present).

• 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.

class dagster_dbt.DbtManifestAssetSelection(*args, **kwargs)[source]¶

Defines a selection of assets from a parsed dbt manifest.json file and a dbt-syntax selection string.

Parameters:

• manifest_json (Mapping[str, Any]) – The parsed manifest.json file from your dbt project. Must provide either this argument or manifest_json_path.

• manifest_json_path – (Optional[str]): The path to a manifest.json file representing the current state of your dbt project. Must provide either this argument or manifest_json.

• select (str) – A dbt-syntax selection string, e.g. tag:foo or config.materialized:table.

• exclude (str) – A dbt-syntax exclude string. Defaults to “”.

• resource_types (Sequence[str]) – The resource types to select. Defaults to [“model”].

• node_info_to_asset_key (Callable[[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. If you pass in a custom function to load_assets_from_dbt_manifest, you must also pass in the same function here.

• state_path – (Optional[str]): The path to a folder containing the manifest.json file representing the previous state of your dbt project. Providing this path will allow you to select dbt assets using the state: selector. To learn more, see the [dbt docs](https://docs.getdbt.com/reference/node-selection/methods#the-state-method).

Example

my_dbt_assets = load_assets_from_dbt_manifest(
    manifest_json,
    node_info_to_asset_key=my_node_info_to_asset_key_fn,
)

# This will select all assets that have the tag "foo" and are in the path "marts/finance"
my_selection = DbtManifestAssetSelection(
    manifest_json,
    select="tag:foo,path:marts/finance",
    node_info_to_asset_key=my_node_info_to_asset_key_fn,
)

# This will retrieve the asset keys according to the selection
selected_asset_keys = my_selection.resolve(my_dbt_assets)

Ops (dbt Core)¶

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.

dagster_dbt.dbt_run_op = <dagster._core.definitions.op_definition.OpDefinition object>[source]¶

Config Schema:

yield_materializations (Bool, optional):

If True, materializations corresponding to the results of the dbt operation will be yielded when the op executes. Default: True

Default Value: True

asset_key_prefix (List[String], optional):

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()

dagster_dbt.dbt_compile_op(context)[source]¶

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()

dagster_dbt.dbt_ls_op(context)[source]¶

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()

dagster_dbt.dbt_test_op(context)[source]¶

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()

dagster_dbt.dbt_snapshot_op(context)[source]¶

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()

dagster_dbt.dbt_seed_op(context)[source]¶

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()

dagster_dbt.dbt_docs_generate_op(context)[source]¶

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()

Resources (dbt Core)¶

Assets (dbt Cloud)¶

dagster_dbt.load_assets_from_dbt_cloud_job(dbt_cloud, job_id, node_info_to_asset_key=<function _get_node_asset_key>, node_info_to_group_fn=<function _get_node_group_name>, node_info_to_freshness_policy_fn=<function _get_node_freshness_policy>, partitions_def=None, partition_key_to_vars_fn=None)[source]¶

Loads a set of dbt models, managed by a dbt Cloud job, into Dagster assets. In order to determine the set of dbt models, the project is compiled to generate the necessary artifacts that define the dbt models and their dependencies.

One Dagster asset is created for each dbt model.

Parameters:

• dbt_cloud (ResourceDefinition) – The dbt Cloud resource to use to connect to the dbt Cloud API.

• job_id (int) – The ID of the dbt Cloud job to load assets from.

• 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])

• 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.

• node_info_to_freshness_policy_fn (Dict[str, Any] -> Optional[FreshnessPolicy]) – A function that takes a dictionary of dbt node info and optionally returns a FreshnessPolicy that should be applied to this node. By default, freshness policies will be created from config applied to dbt models, i.e.: dagster_freshness_policy={“maximum_lag_minutes”: 60, “cron_schedule”: “0 9 * * *”} will result in that model being assigned FreshnessPolicy(maximum_lag_minutes=60, cron_schedule=”0 9 * * *”)

• node_info_to_definition_metadata_fn (Dict[str, Any] -> Optional[Dict[str, MetadataUserInput]]) – A function that takes a dictionary of dbt node info and optionally returns a dictionary of metadata to be attached to the corresponding definition. This is added to the default metadata assigned to the node, which consists of the node’s schema (if present).

• 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”})

Returns:

A definition for the loaded assets.

Return type:

CacheableAssetsDefinition

Examples

from dagster import repository
from dagster_dbt import dbt_cloud_resource, load_assets_from_dbt_cloud_job

DBT_CLOUD_JOB_ID = 1234

dbt_cloud = dbt_cloud_resource.configured(
    {
        "auth_token": {"env": "DBT_CLOUD_API_TOKEN"},
        "account_id": {"env": "DBT_CLOUD_ACCOUNT_ID"},
    }
)

dbt_cloud_assets = load_assets_from_dbt_cloud_job(
    dbt_cloud=dbt_cloud, job_id=DBT_CLOUD_JOB_ID
)


@repository
def dbt_cloud_sandbox():
    return [dbt_cloud_assets]

Ops (dbt Cloud)¶

dagster_dbt.dbt_cloud_run_op = <dagster._core.definitions.op_definition.OpDefinition object>[source]¶

Config Schema:

job_id (Int):

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}/

poll_interval (Float, optional):

The time (in seconds) that will be waited between successive polls.

Default Value: 10

poll_timeout (Union[Float, None], optional):

The maximum time that will waited before this operation is timed out. By default, this will never time out.

Default Value: None

yield_materializations (Bool, optional):

If True, materializations corresponding to the results of the dbt operation will be yielded when the op executes.

Default Value: True

asset_key_prefix (List[String], optional):

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:

job_id (int)

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}/

poll_interval (float)

The time (in seconds) that will be waited between successive polls. Defaults to 10.

poll_timeout (float)

The maximum time (in seconds) that will waited before this operation is timed out. By default, this will never time out.

yield_materializations (bool)

If True, materializations corresponding to the results of the dbt operation will be yielded when the solid executes. Defaults to True.

rasset_key_prefix (float)

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()

Resources (dbt Cloud)¶

dagster_dbt.DbtCloudResourceV2¶

alias of DbtCloudResource

dagster_dbt.dbt_cloud_resource ResourceDefinition[source]¶

Config Schema:

auth_token (dagster.StringSource):

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.

account_id (dagster.IntSource):

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/.

disable_schedule_on_trigger (Bool, optional):

Specifies if you would like any job that is triggered using this resource to automatically disable its schedule.

Default Value: True

request_max_retries (Int, optional):

The maximum number of times requests to the dbt Cloud API should be retried before failing.

Default Value: 3

request_retry_delay (Float, optional):

Time (in seconds) to wait between each request retry.

Default Value: 0.25

dbt_cloud_host (dagster.StringSource, optional):

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():
    ...