feat: get execute dbt model from smas template

pipelines/cadunico/ingest_raw/flows.py

@@ -8,7 +8,6 @@
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
 from prefect.utilities.edges import unmapped
-from prefeitura_rio.core import settings
 from prefeitura_rio.pipelines_utils.custom import Flow
 from prefeitura_rio.pipelines_utils.prefect import (
     task_get_current_flow_run_labels,
@@ -58,7 +57,9 @@
     )
     # Tasks
     project_id = get_project_id_task()
-    materialization_flow_id = task_get_flow_group_id(flow_name=settings.FLOW_NAME_EXECUTE_DBT_MODEL)
+    materialization_flow_id = task_get_flow_group_id(
+        flow_name=constants.FLOW_NAME_EXECUTE_DBT_MODEL_SMAS.value
+    )
     materialization_labels = task_get_current_flow_run_labels()
 
     existing_partitions = get_existing_partitions(

pipelines/cadunico/ingest_raw/utils.py

@@ -792,3 +792,25 @@ def get_dbt_models_to_materialize(
     parameters_list_log = json.dumps(parameters_list_ordered, indent=4)
     log(f"{len(parameters_list_ordered)} TABLES TO MATERIALIZE:\n{parameters_list_log}")
     return parameters_list_ordered
+
+
+def update_local_layout():
+    project_id = "rj-smas"
+    layout_dataset_id = "protecao_social_cadunico"
+    layout_table_id = "layout"
+
+    model_dataset_id = "protecao_social_cadunico"
+    model_table_id = "registro_familia"
+    force_create_models = True
+
+    output_path = f"/tmp/cadunico/layout_parsed/{layout_dataset_id}/{layout_table_id}"
+
+    update_layout_from_storage_and_create_versions_dbt_models(
+        project_id=project_id,
+        layout_dataset_id=layout_dataset_id,
+        layout_table_id=layout_table_id,
+        output_path=output_path,
+        model_dataset_id=model_dataset_id,
+        model_table_id=model_table_id,
+        force_create_models=force_create_models,
+    )

pipelines/templates/run_dbt_model/flows.py

@@ -16,6 +16,8 @@
 from pipelines.constants import constants
 
 templates__run_dbt_model_smas__flow = deepcopy(templates__run_dbt_model__flow)
+templates__run_dbt_model_smas__flow.name = constants.FLOW_NAME_EXECUTE_DBT_MODEL_SMAS
+
 templates__run_dbt_model_smas__flow.state_handlers = [handler_inject_bd_credentials]
 
 templates__run_dbt_model_smas__flow.storage = GCS(constants.GCS_FLOWS_BUCKET.value)