Alterar interval tabelas auxiliares bilhetagem (#525)

* create default materialization flow

* create tasks for default materialization flow

* make generate_execute_schedules more generic

* create bilhetagem materialization flow

* adapt bilhetagem schedules for the new model

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* add run config and storage

* Update utils.py

* fix sub tasks

* fix fetch_dataset_sha run

* add run_date variable to materialization flow

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* remove discord notifications for testing

* add manual date_range / fix flow run name

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix missing table_id logic

* fix empty return

* fix empty return

* add flag_date_range when var_params is blank

* change rename logic when has date variables

* change return values of create_dbt_run_vars

* create dict aux function

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* remove *args from task

* change coalesce task

* fix rename task

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix task order

* add docstrings

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix line too long

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* pre-commit hook

* adjust tasks

* mudar estrutura do flow materializacao

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* adicionar schedule de bilhetagem

* adicionar schedule no flow de materialização

* ajuste nome da coluna de datetime

* ajustar nome coluna

* mudar coluna de data para datetime_transacao

* ajusta variavel date_range manual

* mudar nome parametro de variável dbt

* cria flow de orquestração materialização

* volta notificação do discord

* ajusta wait_flow_run

* mudar query para teste

* reverter query teste

* usar copy no dicionario de variaveis de data

* adjust constant run interval

* remover funcao comentada

* alterar padrão de nome dos flows

* remove imports comentados

* remove schedules nao utilizados

* remove task comentada

* mudar agent para produção

* mudar run interval tabelas auxiliares

* remove tratamento comentado

* ajusta dicionario constante

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Co-authored-by: Rodrigo Cunha <[email protected]>

pipelines/rj_smtr/br_rj_riodejaneiro_bilhetagem/schedules.py

@@ -27,9 +27,7 @@
     dataset_id=constants.BILHETAGEM_DATASET_ID.value,
     secret_path=constants.BILHETAGEM_SECRET_PATH.value,
     source_type=constants.BILHETAGEM_GENERAL_CAPTURE_PARAMS.value["source_type"],
-    runs_interval_minutes=constants.BILHETAGEM_GENERAL_CAPTURE_PARAMS.value[
-        "transacao_runs_interval_minutes"
-    ],
+    runs_interval_minutes=0,
 )
 
 bilhetagem_transacao_schedule = Schedule(clocks=untuple(bilhetagem_transacao_clocks))

pipelines/rj_smtr/constants.py

@@ -183,13 +183,11 @@ class constants(Enum):  # pylint: disable=c0103
         },
         "vpn_url": "http://vpn-jae.mobilidade.rio/",
         "source_type": "api-json",
-        "transacao_runs_interval_minutes": 0,
-        "principal_runs_interval_minutes": 5,
     }
 
     BILHETAGEM_CAPTURE_RUN_INTERVAL = {
         "transacao_run_interval": {"minutes": 1},
-        "principal_run_interval": {"days": 1},
+        "principal_run_interval": {"hours": 1},
     }
 
     BILHETAGEM_TRANSACAO_CAPTURE_PARAMS = {

pipelines/rj_smtr/flows.py

@@ -140,8 +140,6 @@
     flags = Parameter("flags", default=None)
     dbt_vars = Parameter("dbt_vars", default=dict())
 
-    # treated_table_params = treat_dbt_table_params(table_params=table_params)
-
     LABELS = get_current_flow_labels()
     MODE = get_current_flow_mode(LABELS)