I know I can create an upstream dependency to a db...
# ask-community
I know I can create an upstream dependency to a dbt model with
@asset( deps = {AssetKey(["full", "path", "upstream_dbt_model_name"]) })
def *my_custom_asset():*
Is there a way to make
an upstream dependency of all the dbt models which are downstream of the
Hi @Amir Jaber, if I understand you correctly, you can do the following: You have to specify the dagster asset (my_custom_asset) as a source for your dbt model. This is done directly in dbt, in the yaml config (sources config). Here is the documentation that shows you how: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies
Thanks @David Weber, I’ll try this out
Sure, let me know if it solves your issue
Hi @David Weber I seem to be on the right path, as my custom python asset is now in the right place in the DAG visually. However, when I try to materialise it, I get this error:
dagster._core.errors.DagsterInvalidDefinitionError: Asset [‘dagster’, ‘my_source_name’] specifies a schema with its key prefixes [‘dagster’], but schema my_schema_name was also provided via run config. Schema can only be specified one way.
For context, here’s how my dbt assets are imported:
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCliResource):
yield from dbt.cli([“build -f”], context=context).stream()
And how my custom asset is initiated
deps = {
AssetKey([“integration”, “upstream_dbt_model_name”]),
group_name = “python”,
key=get_asset_key_for_source([my_dbt_assets], “dagster”),
def my_source_name(context):
Hi @Amir Jaber, I am definitely not sure, but I think the issue is that you load your source as a Dagster asset. Because you wanted a Dagster asset to be your source model in dbt right? If so, then you don't have to separately load it, as you do in my_source_name But not sure if I am missing something here and if this is connected to the error.
Thanks for your answer @David Weber. I was under the understanding from that docs page that I should create a dbt source which references the output table of
. Then I make sure my_custom_asset is named the same as that source and is connected to it using this syntax. Is there more nuance here that I might be missing?
So if I understood you correctly in your post, you have dagster assets and dbt models. And you want to have the dependency between dagster assets and dbt models - in a sense that dagster assets are upstream (e.g. "before") your dbt models. In that case, you define your last Dagster asset as your dbt source, on which your dbt models depend. You dont have to define the dbt source as a Dagster asset to achieve this. Example: Your last dagster asset:
Copy code
def my_last_dagster_asset():
  # this asset outputs data, which I want to use as a source in dbt
And then in order to define this as a source, you do the following in dbt in your `sources.yml`:
Copy code
version: 2

  - name: my_name
      - name: my_last_dagster_asset
            asset_key: ["my_last_dagster_asset"]
And of course you use this source in your dbt model now:
Copy code
select * from {{ source("dagster", "my_last_dagster_asset") }}
And bam - you have the dependency from your last dagster asset to your first dbt model. There will be no specific "source asset" in dagster, as typically your "source" is the output of your last dagster asset anyway. This is defined here: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies
Thanks @David Weber. I had to figure out some environment variable hackery in order to enable environment separation, but otherwise your write up is exactly what I was looking for!
Amazing @Amir Jaber. Happy to hear that!