Merge branch 'master' into pre-commit-ci-update-config

Author: mergify[bot]

pipelines/rj_smtr/br_rj_riodejaneiro_bilhetagem/flows.py

@@ -238,7 +238,7 @@
 
 bilhetagem_materializacao_gps_validador = set_default_parameters(
     flow=bilhetagem_materializacao_gps_validador,
-    default_parameters=constants.BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_PARAMS.value,
+    default_parameters=constants.BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_GENERAL_PARAMS.value,
 )
 
 bilhetagem_materializacao_gps_validador.state_handlers.append(skip_if_running_handler)
@@ -377,12 +377,11 @@
             flow_name=bilhetagem_materializacao_gps_validador.name,
             project_name=emd_constants.PREFECT_DEFAULT_PROJECT.value,
             labels=LABELS,
-            upstream_tasks=[
-                wait_materializacao_transacao,
-            ],
             parameters={
+                "table_id": constants.BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_TABLE_ID.value,
                 "timestamp": materialize_timestamp,
             },
+            upstream_tasks=[wait_materializacao_transacao],
         )
 
         wait_materializacao_gps_validador = wait_for_flow_run(
@@ -392,6 +391,24 @@
             raise_final_state=True,
         )
 
+        run_materializacao_gps_validador_van = create_flow_run(
+            flow_name=bilhetagem_materializacao_gps_validador.name,
+            project_name=emd_constants.PREFECT_DEFAULT_PROJECT.value,
+            labels=LABELS,
+            parameters={
+                "table_id": constants.BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_VAN_TABLE_ID.value,
+                "timestamp": materialize_timestamp,
+            },
+            upstream_tasks=[wait_materializacao_gps_validador],
+        )
+
+        wait_materializacao_gps_validador_van = wait_for_flow_run(
+            run_materializacao_gps_validador_van,
+            stream_states=True,
+            stream_logs=True,
+            raise_final_state=True,
+        )
+
 bilhetagem_transacao_tratamento.storage = GCS(emd_constants.GCS_FLOWS_BUCKET.value)
 bilhetagem_transacao_tratamento.run_config = KubernetesRun(
     image=emd_constants.DOCKER_IMAGE.value,

pipelines/rj_smtr/constants.py

@@ -547,10 +547,10 @@ class constants(Enum):  # pylint: disable=c0103
         },
     }
 
-    BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_PARAMS = {
+    BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_GENERAL_PARAMS = {
         "dataset_id": BILHETAGEM_DATASET_ID,
-        "table_id": "gps_validador",
         "upstream": True,
+        "downstream": True,
         "exclude": "+operadoras +consorcios",
         "dbt_vars": {
             "date_range": {
@@ -561,6 +561,9 @@ class constants(Enum):  # pylint: disable=c0103
         },
     }
 
+    BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_TABLE_ID = "gps_validador"
+    BILHETAGEM_MATERIALIZACAO_GPS_VALIDADOR_VAN_TABLE_ID = "gps_validador_van"
+
     BILHETAGEM_GENERAL_CAPTURE_DEFAULT_PARAMS = {
         "dataset_id": BILHETAGEM_DATASET_ID,
         "secret_path": BILHETAGEM_SECRET_PATH,