From 409855fb4649b786cac638c6339e1fff40c98f90 Mon Sep 17 00:00:00 2001
From: luutuankiet <luutuankiet.ftu2@gmail.com>
Date: Mon, 26 Feb 2024 15:46:17 +0700
Subject: [PATCH] setup multi asset for job dump motherduck matching new
 dagster version

---
 app/ETL/EL.py                  | 18 ++++++++++++++----
 app/ETL/definitions.py         |  2 +-
 dbt_project/models/_source.yml | 17 ++++-------------
 3 files changed, 19 insertions(+), 18 deletions(-)

diff --git a/app/ETL/EL.py b/app/ETL/EL.py
index 3289cc8..d2a9197 100644
--- a/app/ETL/EL.py
+++ b/app/ETL/EL.py
@@ -1,11 +1,21 @@
 import os
-from dagster import asset,AssetExecutionContext
+from dagster import asset,AssetExecutionContext,AssetOut, multi_asset, load_assets_from_modules
+from dagster_dbt import get_asset_keys_by_output_name_for_source
 import duckdb
 import pandas as pd
 from helper.source_env import raw_path
-
-
-@asset(compute_kind='Python')
+from . import dbt_assets
+
+
+@multi_asset(
+    outs={
+        name: AssetOut(key=asset_key)
+        for name, asset_key in get_asset_keys_by_output_name_for_source(
+            [dbt_assets.ticktick_dbt_assets], "raw_data"
+        ).items()
+    },
+    compute_kind='python'
+)
 def dump_to_motherduck(context: AssetExecutionContext):
     motherduck_token=os.environ.get('motherduck_token')
 
diff --git a/app/ETL/definitions.py b/app/ETL/definitions.py
index 8dffaa3..3ebe3de 100644
--- a/app/ETL/definitions.py
+++ b/app/ETL/definitions.py
@@ -3,7 +3,7 @@
 from dagster_dbt import DbtCliResource
 
 from .constants import DBT_PROJECT_DIR
-from. import EL,dbt_assets
+from . import EL,dbt_assets
 
 all_assets = load_assets_from_modules([EL,dbt_assets])
 ETL_job = define_asset_job("ETL_job",selection=all_assets)
diff --git a/dbt_project/models/_source.yml b/dbt_project/models/_source.yml
index 6a70032..263c94b 100644
--- a/dbt_project/models/_source.yml
+++ b/dbt_project/models/_source.yml
@@ -2,27 +2,18 @@ version: 2
 
 sources:
   - name: raw_data
+    meta:
+      dagster:
+        asset_key : ["dump_to_motherduck"]
     description: A replica of the extracted ticktick data.
     database: ticktick_gtd
     schema: main
     tables:
       - name: tasks_raw
         description: raw tasks data from tickpy
-        meta:
-          dagster:
-            asset_key: ["dump_to_motherduck"]
       - name: lists_raw
         description: raw lists data from tickpy
-        meta:
-          dagster:
-            asset_key: ["dump_to_motherduck"]
       - name: folders_raw
-        description: raw folders data from tickpy
-        meta:
-          dagster:
-            asset_key: ["dump_to_motherduck"]            
+        description: raw folders data from tickpy      
       - name: date_seed
         description: pre-generated dates
-        meta:
-          dagster:
-            asset_key: ["dump_to_motherduck"]