Fix model_name retrieval; get from dbt_node_config

astronomer · Oct 1, 2024 · 72c6164 · 72c6164
1 parent c3c51cb
commit 72c6164
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/cosmos/operators/airflow_async.py b/cosmos/operators/airflow_async.py
@@ -96,7 +96,7 @@ def get_remote_sql(self) -> str:
             return fp.read()  # type: ignore
 
     def drop_table_sql(self) -> None:
-        model_name = self.task_id.split(".")[0]
+        model_name = self.extra_context["dbt_node_config"]["resource_name"]
         sql = f"DROP TABLE IF EXISTS {self.gcp_project}.{self.dataset}.{model_name};"
         hook = BigQueryHook(
             gcp_conn_id=self.gcp_conn_id,
@@ -117,7 +117,7 @@ def execute(self, context: Context) -> Any | None:
             self.drop_table_sql()
 
             sql = self.get_remote_sql()
-            model_name = self.task_id.split(".")[0]
+            model_name = self.extra_context["dbt_node_config"]["resource_name"]
             # prefix explicit create command to create table
             sql = f"CREATE TABLE {self.gcp_project}.{self.dataset}.{model_name} AS  {sql}"