I'm trying to figure out how to feed a dbt asset into an op using the new API (for example, I want to read in a dbt table and send an email based on some calculations):

@dbt_assets(
    manifest=manifest
)
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 defining an I/O manager so the job doesn't know how to load the table

We have APIs for specifying downstream dependencies for certain dbt models: https://github.com/dagster-io/dagster/discussions/14964

I read through that, but all the examples are of downstream assets - and I have a downstream op. Would it be helpful to add my use case to the discussion?

I think what you need is an io manager for the dbt assets. Then in the

load_input

function of the io manager, connect to the database and do a select star from the dbt asset and return the results as a dataframe. Could be pandas or polars. I'd recommend polars though. It's be nice if an io manager like that came out of the box from dagster that uses the underlying dbt adapter to connect to the warehouse and get the results. But at the moment, we don't have that

thanks @Qwame! that was it - I somehow overlooked that the new

@dbt_assets

decorator has an

io_manager_key

argument 🤦