Skip to content

Commit 3df95e3

Browse files
committed
Add asset_key_prefix support for dbt sources with tests and docs
1 parent 93a9d89 commit 3df95e3

10 files changed

Lines changed: 236 additions & 0 deletions

File tree

docs/docs/integrations/libraries/dbt/reference.md

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -309,6 +309,37 @@ sources:
309309
asset_key: ['snowflake', 'jaffle_shop', 'orders']
310310
```
311311
312+
If you want to apply a shared prefix without repeating a full key for each node, you can configure
313+
`meta.dagster.asset_key_prefix` and Dagster will append the dbt node name:
314+
315+
```yaml
316+
models:
317+
jaffle_shop:
318+
+meta:
319+
dagster:
320+
asset_key_prefix: ['snowflake', 'jaffle_shop']
321+
```
322+
323+
For `orders`, this produces the asset key `snowflake/jaffle_shop/orders`.
324+
325+
For dbt sources, the source name is preserved between the prefix and the table name:
326+
327+
```yaml
328+
sources:
329+
- name: jaffle_shop
330+
meta:
331+
dagster:
332+
asset_key_prefix: ['snowflake']
333+
```
334+
335+
For a table `orders` in source `jaffle_shop`, this produces the asset key `snowflake/jaffle_shop/orders`.
336+
337+
When both are configured, Dagster uses this precedence:
338+
339+
1. `meta.dagster.asset_key`
340+
2. `meta.dagster.asset_key_prefix` + `[node_name]` (models/snapshots) or `[source_name/table_name]` (sources)
341+
3. Default Dagster dbt key generation
342+
312343
Alternatively, to override the asset key generation for all dbt nodes in your dbt project, you can create a custom <PyObject section="libraries" integration="dbt" module="dagster_dbt" object="DagsterDbtTranslator" /> and implement <PyObject section="libraries" integration="dbt" module="dagster_dbt" object="DagsterDbtTranslator.get_asset_key" />. The following example adds a `snowflake` prefix to the default generated asset key:
313344

314345
<CodeExample

python_modules/libraries/dagster-dbt/dagster_dbt/asset_utils.py

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -644,12 +644,17 @@ def default_asset_key_fn(dbt_resource_props: Mapping[str, Any]) -> AssetKey:
644644
dbt sources: a dbt source's key is the union of its source name and its table name
645645
dbt models: a dbt model's key is the union of its model name and any schema configured on
646646
the model itself.
647+
If an asset key prefix is configured in its metadata, it is prepended
648+
to the default asset key. The prefix supports project or folder level
649+
configuration in `dbt_project.yml`.
647650
"""
648651
dbt_meta = dbt_resource_props.get("config", {}).get("meta", {}) or dbt_resource_props.get(
649652
"meta", {}
650653
)
651654
dagster_metadata = dbt_meta.get("dagster", {})
652655
asset_key_config = dagster_metadata.get("asset_key", [])
656+
asset_key_prefix_config = dagster_metadata.get("asset_key_prefix", [])
657+
653658
if asset_key_config:
654659
return AssetKey(asset_key_config)
655660

@@ -664,6 +669,14 @@ def default_asset_key_fn(dbt_resource_props: Mapping[str, Any]) -> AssetKey:
664669
else:
665670
components = [dbt_resource_props["name"]]
666671

672+
if asset_key_prefix_config:
673+
prefix = (
674+
[asset_key_prefix_config]
675+
if isinstance(asset_key_prefix_config, str)
676+
else list(asset_key_prefix_config)
677+
)
678+
components = [*prefix, *components]
679+
667680
return AssetKey(components)
668681

669682

python_modules/libraries/dagster-dbt/dagster_dbt_tests/conftest.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
from dagster_dbt_tests.dbt_projects import (
1212
test_asset_checks_path,
1313
test_asset_key_exceptions_path,
14+
test_asset_key_prefix_path,
1415
test_dagster_dbt_mixed_freshness_path,
1516
test_dbt_alias_path,
1617
test_dbt_functions_path,
@@ -115,6 +116,11 @@ def test_asset_key_exceptions_manifest_fixture() -> dict[str, Any]:
115116
return _create_dbt_invocation(test_asset_key_exceptions_path).get_artifact("manifest.json")
116117

117118

119+
@pytest.fixture(name="test_asset_key_prefix_manifest", scope="session")
120+
def test_asset_key_prefix_manifest_fixture() -> dict[str, Any]:
121+
return _create_dbt_invocation(test_asset_key_prefix_path).get_artifact("manifest.json")
122+
123+
118124
@pytest.fixture(name="test_dbt_alias_manifest", scope="session")
119125
def test_dbt_alias_manifest_fixture() -> dict[str, Any]:
120126
return _create_dbt_invocation(test_dbt_alias_path).get_artifact("manifest.json")
Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
from typing import TYPE_CHECKING, Any
2+
3+
import pytest
4+
from dagster import AssetKey
5+
from dagster_dbt import build_dbt_asset_specs, dbt_assets
6+
from dagster_dbt.asset_utils import default_asset_key_fn
7+
8+
if TYPE_CHECKING:
9+
from collections.abc import Mapping
10+
11+
12+
def test_default_asset_key_fn_asset_key_takes_precedence_over_prefix() -> None:
13+
dbt_resource_props: Mapping[str, Any] = {
14+
"name": "orders",
15+
"resource_type": "model",
16+
"config": {
17+
"meta": {
18+
"dagster": {
19+
"asset_key": ["warehouse", "jaffle_shop", "orders"],
20+
"asset_key_prefix": ["ignored", "prefix"],
21+
}
22+
}
23+
},
24+
}
25+
26+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(
27+
["warehouse", "jaffle_shop", "orders"]
28+
)
29+
30+
31+
@pytest.mark.parametrize(
32+
"resource_type,name",
33+
[
34+
("model", "orders"),
35+
("snapshot", "orders_snapshot"),
36+
],
37+
)
38+
def test_default_asset_key_fn_asset_key_prefix_for_supported_resource_types(
39+
resource_type: str, name: str
40+
) -> None:
41+
dbt_resource_props = {
42+
"name": name,
43+
"resource_type": resource_type,
44+
"source_name": "jaffle_shop",
45+
"config": {"meta": {"dagster": {"asset_key_prefix": ["snowflake", "jaffle_shop"]}}},
46+
}
47+
48+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(["snowflake", "jaffle_shop", name])
49+
50+
51+
def test_default_asset_key_fn_asset_key_prefix_for_source() -> None:
52+
# For sources, the key is: prefix + source_name + table_name
53+
dbt_resource_props = {
54+
"name": "raw_orders",
55+
"resource_type": "source",
56+
"source_name": "jaffle_shop",
57+
"config": {"meta": {"dagster": {"asset_key_prefix": ["snowflake"]}}},
58+
}
59+
60+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(
61+
["snowflake", "jaffle_shop", "raw_orders"]
62+
)
63+
64+
65+
def test_default_asset_key_fn_asset_key_prefix_uses_alias_for_versioned_models() -> None:
66+
dbt_resource_props = {
67+
"name": "orders",
68+
"alias": "orders_v1",
69+
"version": 1,
70+
"resource_type": "model",
71+
"config": {"meta": {"dagster": {"asset_key_prefix": ["snowflake", "jaffle_shop"]}}},
72+
}
73+
74+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(
75+
["snowflake", "jaffle_shop", "orders_v1"]
76+
)
77+
78+
79+
def test_default_asset_key_fn_asset_key_prefix_from_meta_path() -> None:
80+
# Sources read asset_key_prefix from top-level meta, not config.meta
81+
dbt_resource_props = {
82+
"name": "orders",
83+
"resource_type": "source",
84+
"source_name": "jaffle_shop",
85+
"config": {},
86+
"meta": {"dagster": {"asset_key_prefix": ["raw"]}},
87+
}
88+
89+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(["raw", "jaffle_shop", "orders"])
90+
91+
92+
def test_default_asset_key_fn_empty_asset_key_prefix_uses_node_name() -> None:
93+
dbt_resource_props = {
94+
"name": "orders",
95+
"resource_type": "model",
96+
"config": {"meta": {"dagster": {"asset_key_prefix": []}}},
97+
}
98+
99+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(["orders"])
100+
101+
102+
def test_default_asset_key_fn_asset_key_prefix_accepts_string() -> None:
103+
dbt_resource_props = {
104+
"name": "orders",
105+
"resource_type": "model",
106+
"config": {"meta": {"dagster": {"asset_key_prefix": "snowflake"}}},
107+
}
108+
109+
assert default_asset_key_fn(dbt_resource_props) == AssetKey(["snowflake", "orders"])
110+
111+
112+
def test_dbt_meta_asset_key_prefix_with_inherited_config(
113+
test_asset_key_prefix_manifest: dict[str, Any],
114+
) -> None:
115+
expected_specs_by_key = {
116+
spec.key: spec for spec in build_dbt_asset_specs(manifest=test_asset_key_prefix_manifest)
117+
}
118+
119+
@dbt_assets(manifest=test_asset_key_prefix_manifest)
120+
def my_dbt_assets(): ...
121+
122+
assert AssetKey(["warehouse", "jaffle_shop", "orders_from_source"]) in my_dbt_assets.keys
123+
assert AssetKey(["override", "explicit_asset_key"]) in my_dbt_assets.keys
124+
assert AssetKey(["warehouse", "jaffle_shop", "explicit_asset_key"]) not in my_dbt_assets.keys
125+
126+
assert AssetKey(["staging", "raw", "orders"]) in set(my_dbt_assets.keys_by_input_name.values())
127+
128+
assert AssetKey(["warehouse", "jaffle_shop", "orders_from_source"]) in expected_specs_by_key
129+
assert AssetKey(["override", "explicit_asset_key"]) in expected_specs_by_key

python_modules/libraries/dagster-dbt/dagster_dbt_tests/dbt_projects/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,3 +31,4 @@
3131
test_dagster_dbt_mixed_freshness_path = projects_path.joinpath("test_dagster_dbt_mixed_freshness")
3232
test_jaffle_with_profile_vars_path = projects_path.joinpath("test_jaffle_with_profile_vars")
3333
test_dependencies_path = projects_path.joinpath("test_dagster_dbt_dependencies")
34+
test_asset_key_prefix_path = projects_path.joinpath("test_dagster_asset_key_prefix")
Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
name: "test_dagster_asset_key_prefix"
2+
3+
config-version: 2
4+
version: "0.1"
5+
6+
profile: "jaffle_shop"
7+
8+
model-paths: ["models"]
9+
seed-paths: ["seeds"]
10+
test-paths: ["tests"]
11+
analysis-paths: ["analysis"]
12+
macro-paths: ["macros"]
13+
14+
target-path: "target"
15+
clean-targets:
16+
- "target"
17+
- "dbt_modules"
18+
- "logs"
19+
20+
require-dbt-version: [">=1.0.0", "<2.0.0"]
21+
22+
models:
23+
test_dagster_asset_key_prefix:
24+
+meta:
25+
dagster:
26+
asset_key_prefix: ["warehouse", "jaffle_shop"]
27+
28+
sources:
29+
test_dagster_asset_key_prefix:
30+
+meta:
31+
dagster:
32+
asset_key_prefix: ["staging"]
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
select 1 as id
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
select * from {{ source('raw', 'orders') }}
Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
version: 2
2+
3+
sources:
4+
- name: raw
5+
tables:
6+
- name: orders
7+
8+
models:
9+
- name: orders_from_source
10+
11+
- name: explicit_asset_key
12+
config:
13+
meta:
14+
dagster:
15+
asset_key: ["override", "explicit_asset_key"]
Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
jaffle_shop:
2+
target: dev
3+
outputs:
4+
dev:
5+
type: duckdb
6+
path: "{{ env_var('DAGSTER_DBT_PYTEST_XDIST_DUCKDB_DBFILE_PATH') }}"
7+
threads: 24

0 commit comments

Comments
 (0)