[RFC] Improved APIs for defining dbt asset dependencies

[clairelin135]

This discussion details new experimental APIs meant to improve the ergonomics of defining python dependencies upstream and downstream of dbt assets. This is a continuation of the changes described in this prior discussion: #14477.

Please comment with any questions or feedback! Feedback here will help to inform future API changes.

Python assets upstream of dbt assets

Before 0.19.12
When a dbt asset has an upstream asset, it is defined as a source in a YAML file. In Dagster, you can add a definition for this source by explicitly specifying the asset key. When the source name or table name changes in dbt, this key would have to be updated in the asset definition:

sources:
  - name: raw_data
    tables:
      - name: orders

@asset(key_prefix="raw_data")
def orders() -> pd.DataFrame:
    ...

After 0.19.12
A new DbtManifest.get_asset_key_for_source method fetches the asset key of a given source. This allows users to update source names or table names in dbt without updating the asset definition:

manifest = DbtManifest.read(path=MANIFEST_PATH)

@asset(key=manifest.get_asset_key_for_source("raw_data"))  # AssetKey(["raw_data", "orders"])
def orders() -> pd.DataFrame:
    ...

If the source contains multiple tables that are created from a single step, users can call the DbtManifest.get_asset_keys_by_output_name_for_source method to define a @multi_asset. This method returns a mapping of asset key by output name, allowing users to specify additional fields on the AssetOut:

@multi_asset(
    outs={
        name: AssetOut(key=asset_key)
        for name, asset_key in manifest.get_asset_keys_by_output_name_for_source(
            "source_with_multiple_tables"
        ).items()
    }
)
def my_multi_asset(context):
    output_names = list(context.selected_output_names)
    ...

Python assets downstream of dbt assets

Before 0.19.12
Downstream assets are defined by explicitly setting the asset key to match the upstream dbt model's key. This requires changing the key when the upstream model changes (i.e. the schema or model name is changed):

@asset(
    non_argument_deps={AssetKey(["dbt_schema", "order_summary"])},
)
def downstream_asset() -> Any:
    ...

After 0.19.12
Users can specify an upstream asset key in the @asset decorator by calling DbtManifest.get_asset_key_for_model:

manifest = DbtManifest.read(path=MANIFEST_PATH)

@asset(non_argument_deps={manifest.get_asset_key_for_model("order_summary")})
def downstream_asset() -> Any:
    ...

[malow106]

Hi, nice feature ! One question : as I see a manifest path input, is this possible with load_asset_from_dbt_project ?

[clairelin135]

You can load assets from a dbt manifest via the load_assets_from_dbt_manifest method, which pretty much functions the same as load_assets_from_dbt_project.

The DbtManifest methods introduced in this discussion are compatible with the load_assets... methods, though the experience will be more natural when using the experimental @dbt_assets method that we recently released. Manipulating DbtManifest will be a first class experience in the experimental APIs.

[geoHeil]

So does this mean that the python (downstream) process no longer needs to be aware of prefixes to normal or source assets? Rather given the upstream manifest reference and the @assets anifest.get_asset_key_for_source("raw_data")) defined earlier automatically reference the right asset keys?

[clairelin135]

The manifest methods get the asset key as defined in the dbt manifest.

But if you added a custom key prefix to upstream dbt assets outside of dbt, you would have to update the returned key with whatever key prefix you added:

dbt_assets = load_assets_from_modules(modules, key_prefix="foo")

@asset(non_argument_deps={manifest.get_asset_key_for_model("order_summary").with_prefix("foo")})
def downstream_asset() -> Any:
    ...

[cdchan]

I think it would be helpful to have an example with a downstream op. (For example, if you have a dbt model that calculates daily revenue as part of the nightly dbt run, and later in the day you want to send an email with yesterday's revenue without re-running dbt.)

Right now I'm doing:

@dbt_assets(
    manifest=manifest,
    io_manager_key="my_data_warehouse_io_manager",
)
def all_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    ...

@job
def my_job():
    my_op(all_dbt_assets.to_source_asset(key=manifest.get_asset_key_for_model(model_name="my_dbt_model")))

but I'm not sure if this is the right idiom. It also doesn't show the dbt asset in the dagit UI graph of the job, but I don't know if it's supposed to (would be neat though)

[clairelin135]

@cdchan Here's an example of how to define an asset downstream of a dbt asset: https://docs.dagster.io/integrations/dbt/reference#downstream-dependencies