diff --git a/pipelines/cadunico/ingest_raw/flows.py b/pipelines/cadunico/ingest_raw/flows.py index 797eb14..f904a51 100644 --- a/pipelines/cadunico/ingest_raw/flows.py +++ b/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( diff --git a/pipelines/cadunico/ingest_raw/utils.py b/pipelines/cadunico/ingest_raw/utils.py index 7825898..c42d3b0 100644 --- a/pipelines/cadunico/ingest_raw/utils.py +++ b/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, + ) diff --git a/pipelines/templates/run_dbt_model/flows.py b/pipelines/templates/run_dbt_model/flows.py index 71b4194..22f23c3 100644 --- a/pipelines/templates/run_dbt_model/flows.py +++ b/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)