dbt_assets yields multiple outputs to one op

[kennycontreras]

Hello everyone 👋

I want to create a job that has one @dbt_assets and one @op as downstream.
The idea is to create an execution order dependency between them, not passing any data.

I've tried multiple ways without luck. I'm not sure if I'm missing something.

Here's a code example:

@dbt_assets(
    manifest=manifest,
    select="my_dbt_project.models.assets"
)
def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build", "--target", "dev"], context=context).stream()

@op(ins={"start": In(Nothing)})
def downstream_op(context):
    pass

@job
def my_dbt_job():
    downstream_op(start=my_dbt_assets())

The error I constantly get is

received a tuple of multiple outputs for input "start" (passed by keyword) in op invocation downstream_op. Must pass individual output, available from tuple

Any suggestions are more than welcome.
Thanks!

[garethbrickman]

I think this is not technically doable as the @dbt_assets decorator yields multiple outputs, and you're trying to pass these directly to an @op that expects a single input.

[kennycontreras]

Thanks for the info. @garethbrickman I managed to create the dependency with assets and ops. I'll add a full example of it.

[kennycontreras]

I managed to create the dependency using ops and also with assets. I'll add a few examples of how to achieve this.

• ops:

@dbt_assets(
    manifest=manifest,
    select="my_dbt_project.models.assets"
)
def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build", "--target", "dev"], context=context).stream()

@op(ins={"start": In(Nothing)})
def downstream_op(context):
    pass

@job()
def my_dbt_job():
    dbt_assets = my_dbt_assets()
    assets = [asset for asset in dbt_assets]
    downstream_op(assets)

With assets, it is nicer because we have the full potential of the assets graph, materialization, etc.

• assets

@dbt_assets(
    manifest=manifest,
    select="my_dbt_project.models.assets"
)
def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build", "--target", "dev"], context=context).stream()

@asset(deps=my_dbt_assets.keys)
def downstream_asset(context: AssetExecutionContext):
    pass 

my_dbt_job = define_asset_job(
    name="my_dbt_job",
    selection=[my_dbt_assets, downstream_asset],
)