How to achieve that dbt assets depends on python assets?

[jordunar]

As long as I have been working on dbt on Dagster I always have faced problems where the dbt assets where loaded the first ones in the DAG and then the python assets ones were loaded. For example, an structure like this:

@dbt_assets(manifest=dbt_manifest_path, select = 'tag:fook') #select='path:models/example/' , select = 'tag:fook'
def dbt_project_test_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


@asset(compute_kind="python", deps=get_asset_key_for_model([dbt_project_test_dbt_assets], "test_dagster_migration_1"))
def asset_prueba():
    lista = ["a", "b", "c"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )

The problem is that I want to do something the other way; I firstly want to create some assets with Python and only then, when these assets are created, execute the dbt models. What I would want to know if there is anything similar as the deps option from the @assets decorators but with the @dbt_assets ones.

Thanks in advanced

[rexledesma]

If you model your upstream dependencies as dbt sources, you can create assets for your dbt sources. These assets will execute before your dbt assets.

See the docs for details: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies

[jordunar]

Okey, that answer my question but my problem it's not exactly this one. I'm going to explain you my problem: what I am doing right now is migrating to Dagster some processes that were executed before using GCP workflow. Those processes were executed independently (we had one that import some images, another one that used the result of those imports and did operations with them, etc.) as parts of a bigger pipeline. The thing is that we have a dbt project where the models folder is divided in subfolders, and, for each one of the workflows executed we used different subfolders. For example, for workflow 1 we firstly executed a dbt run command on the subfolder 1 and then launched a dataflow job, and for workflow 2 we firstly executed a dbt run command on the subfolder 2 and then launched two different dataflow jobs. This way, as you can see, we had an amount of steps that were executed sequentially and independently.

What I want now is to merge all this workflows in just one DAG using Dagster. Therefore, what I want to achieve is connecting the last step of workflow 1 (which is launching a dataflow job and can be done with a dagster asset) with the first step of workflow 2 (which is the launching of some dbt models). Anyway, this connection is not a connection that looks for the first step of wf 2 uses the information of the last step of wf1; it's just a way of indicating Dagster the order of execution of the things. I have tried to use what you've sent me as a source but, as you can imagine, the last asset is not a table or any source of data (is the launching of a dataflow job) and when I materialize it it fails.

What I want to know if it's possible to do is something like this:

@asset(deps=[launch_test_dataflow], compute_kind="python")
def asset_test1():
    lista = ["a", "b", "c"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )


@asset(deps=[asset_test1], compute_kind="python")
def asset_test2():
    lista = ["d", "e", "f"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )

, but the second asset, instead of being a python asset, being a dbt asset. As you can see the second asset here does not use the information of the first asset; this deps argument just indicates us that it needs to be executed only when the first one is executed.

Thanks in advanced,

Javier

[bkimjin]

@jordunar were you able to find a solution?