[dagster-dbt][refactor] Use AssetOut.from_spec() to create AssetOuts …

…for dbt_assets decorator
dagster-io · Dec 20, 2024 · 03c57d7 · 03c57d7
1 parent 234e10c
commit 03c57d7
Showing 1 changed file with 10 additions and 4 deletions.
diff --git a/python_modules/libraries/dagster-dbt/dagster_dbt/asset_utils.py b/python_modules/libraries/dagster-dbt/dagster_dbt/asset_utils.py
@@ -25,6 +25,7 @@
     AssetOut,
     AssetsDefinition,
     AssetSelection,
+    AssetSpec,
     AutoMaterializePolicy,
     AutomationCondition,
     DagsterInvalidDefinitionError,
@@ -807,12 +808,9 @@ def build_dbt_multi_asset_args(
                 project=project,
             )
 
-        outs[output_name] = AssetOut(
+        spec = AssetSpec(
             key=asset_key,
-            dagster_type=Nothing,
-            io_manager_key=io_manager_key,
             description=dagster_dbt_translator.get_description(dbt_resource_props),
-            is_required=False,
             metadata=metadata,
             owners=dagster_dbt_translator.get_owners(
                 {
@@ -832,6 +830,14 @@ def build_dbt_multi_asset_args(
                 dbt_resource_props
             ),
         )
+        if io_manager_key:
+            spec = spec.with_io_manager_key(io_manager_key)
+
+        outs[output_name] = AssetOut.from_spec(
+            spec=spec,
+            dagster_type=Nothing,
+            is_required=False,
+        )
 
         test_unique_ids = [
             child_unique_id