[fix] br_bd_siga_o_dinheiro

pipelines/datasets/br_jota/flows.py

@@ -2,6 +2,7 @@
 """
 Flows for br_jota
 """
+
 from datetime import timedelta
 
 from prefect import Parameter
@@ -15,144 +16,60 @@
 from pipelines.utils.execute_dbt_model.constants import constants as dump_db_constants
 from pipelines.utils.tasks import get_current_flow_labels
 
-with Flow(
- name="br_jota.eleicao_perfil_candidato_2022", code_owners=["lauris"]
-) as eleicao_perfil_candidato_2022:
- dataset_id = Parameter("dataset_id", default="br_jota", required=True)
- table_id = Parameter(
- "table_id", default="eleicao_perfil_candidato_2022", required=True
- )
- materialization_mode = Parameter(
- "materialization_mode", default="prod", required=False
- )
- dbt_alias = Parameter("dbt_alias", default=False, required=False)
-
- current_flow_labels = get_current_flow_labels()
- materialization_flow = create_flow_run(
- flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
- project_name=constants.PREFECT_DEFAULT_PROJECT.value,
- parameters={
- "dataset_id": dataset_id,
- "table_id": table_id,
- "mode": materialization_mode,
- "dbt_alias": dbt_alias,
- },
- labels=current_flow_labels,
- run_name=f"Materialize {dataset_id}.{table_id}",
- )
-
- wait_for_materialization = wait_for_flow_run(
- materialization_flow,
- stream_states=True,
- stream_logs=True,
- raise_final_state=True,
- )
- wait_for_materialization.max_retries = (
- dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
- )
- wait_for_materialization.retry_delay = timedelta(
- seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
- )
-
-eleicao_perfil_candidato_2022.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
-eleicao_perfil_candidato_2022.run_config = KubernetesRun(
- image=constants.DOCKER_IMAGE.value
+from pipelines.datasets.br_jota.tasks import (
+ get_table_ids
 )
-eleicao_perfil_candidato_2022.schedule = None
 
+from pipelines.datasets.br_jota.schedules import (
+ schedule_br_jota_2024
+)
 
 with Flow(
- name="br_jota.eleicao_prestacao_contas_candidato_2022", code_owners=["lauris"]
-) as eleicao_prestacao_contas_candidato_2022:
+ name="BD template - br_jota", code_owners=["luiz"]
+) as br_jota_2024:
+
  dataset_id = Parameter("dataset_id", default="br_jota", required=True)
- table_id = Parameter(
- "table_id", default="eleicao_prestacao_contas_candidato_2022", required=True
- )
+
  materialization_mode = Parameter(
  "materialization_mode", default="dev", required=False
  )
- dbt_alias = Parameter("dbt_alias", default=False, required=False)
-
- current_flow_labels = get_current_flow_labels()
- materialization_flow = create_flow_run(
- flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
- project_name=constants.PREFECT_DEFAULT_PROJECT.value,
- parameters={
- "dataset_id": dataset_id,
- "table_id": table_id,
- "mode": materialization_mode,
- "dbt_alias": dbt_alias,
- },
- labels=current_flow_labels,
- run_name=f"Materialize {dataset_id}.{table_id}",
- )
-
- wait_for_materialization = wait_for_flow_run(
- materialization_flow,
- stream_states=True,
- stream_logs=True,
- raise_final_state=True,
- )
- wait_for_materialization.max_retries = (
- dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
- )
- wait_for_materialization.retry_delay = timedelta(
- seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
- )
-
-eleicao_prestacao_contas_candidato_2022.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
-eleicao_prestacao_contas_candidato_2022.run_config = KubernetesRun(
- image=constants.DOCKER_IMAGE.value
-)
-eleicao_prestacao_contas_candidato_2022.schedule = None
+ dbt_alias = Parameter("dbt_alias", default=True, required=False)
 
+ table_ids = get_table_ids()
 
-with Flow(
- name="br_jota.eleicao_prestacao_contas_candidato_origem_2022",
- code_owners=["lauris"],
-) as eleicao_prestacao_contas_candidato_origem_2022:
- dataset_id = Parameter("dataset_id", default="br_jota", required=True)
- table_id = Parameter(
- "table_id",
- default="eleicao_prestacao_contas_candidato_origem_2022",
- required=True,
- )
- materialization_mode = Parameter(
- "materialization_mode", default="prod", required=False
- )
- dbt_alias = Parameter("dbt_alias", default=False, required=False)
+ for n, table_id in enumerate(table_ids):
 
- current_flow_labels = get_current_flow_labels()
- materialization_flow = create_flow_run(
+  current_flow_labels = get_current_flow_labels(upstream_tasks=[wait_for_materialization] if n > 0 else None)
+  materialization_flow = create_flow_run(
  flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
  project_name=constants.PREFECT_DEFAULT_PROJECT.value,
  parameters={
  "dataset_id": dataset_id,
  "table_id": table_id,
  "mode": materialization_mode,
  "dbt_alias": dbt_alias,
+ "download_csv_file": False
+
  },
  labels=current_flow_labels,
  run_name=f"Materialize {dataset_id}.{table_id}",
- )
-
- wait_for_materialization = wait_for_flow_run(
+ )
+ wait_for_materialization = wait_for_flow_run(
  materialization_flow,
  stream_states=True,
  stream_logs=True,
  raise_final_state=True,
- )
- wait_for_materialization.max_retries = (
+  )
+  wait_for_materialization.max_retries = (
  dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
- )
- wait_for_materialization.retry_delay = timedelta(
+  )
+  wait_for_materialization.retry_delay = timedelta(
  seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
- )
+  )
 
-eleicao_prestacao_contas_candidato_origem_2022.storage = GCS(
- constants.GCS_FLOWS_BUCKET.value
-)
-eleicao_prestacao_contas_candidato_origem_2022.run_config = KubernetesRun(
+
+br_jota_2024.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
+br_jota_2024.run_config = KubernetesRun(
  image=constants.DOCKER_IMAGE.value
 )
-eleicao_prestacao_contas_candidato_origem_2022.schedule = None
+br_jota_2024.schedule = schedule_br_jota_2024

pipelines/datasets/br_jota/schedules.py

@@ -3,66 +3,27 @@
 Schedules for br_jota
 """
 
-from datetime import datetime, timedelta
+from datetime import datetime
 
-from prefect.schedules import Schedule, filters
-from prefect.schedules.clocks import IntervalClock
+from prefect.schedules import Schedule
+from prefect.schedules.clocks import CronClock
 
 from pipelines.constants import constants
 
-schedule_candidatos = Schedule(
- clocks=[
- IntervalClock(
- interval=timedelta(days=1),
- start_date=datetime(2021, 1, 1, 10, 45),
- labels=[
- constants.BASEDOSDADOS_PERGUNTAS_AGENT_LABEL.value,
- ],
- parameter_defaults={
- "dataset_id": "br_jota",
- "table_id": "eleicao_perfil_candidato_2022",
- "materialization_mode": "dev",
- "dbt_alias": False,
- },
- ),
- ],
- filters=[filters.is_weekday],
-)
-
-schedule_contas_candidato = Schedule(
- clocks=[
- IntervalClock(
- interval=timedelta(days=1),
- start_date=datetime(2021, 1, 1, 10, 50),
- labels=[
- constants.BASEDOSDADOS_PERGUNTAS_AGENT_LABEL.value,
- ],
- parameter_defaults={
- "dataset_id": "br_jota",
- "table_id": "eleicao_prestacao_contas_candidato_2022",
- "materialization_mode": "dev",
- "dbt_alias": False,
- },
- ),
- ],
- filters=[filters.is_weekday],
-)
 
-schedule_contas_candidato_origem = Schedule(
+schedule_br_jota_2024 = Schedule(
  clocks=[
- IntervalClock(
- interval=timedelta(days=1),
- start_date=datetime(2021, 1, 1, 10, 55),
+ CronClock(
+ cron="30 4 * * *",
+ start_date=datetime(2024, 9, 1, 0, 0),
  labels=[
  constants.BASEDOSDADOS_PERGUNTAS_AGENT_LABEL.value,
  ],
  parameter_defaults={
  "dataset_id": "br_jota",
- "table_id": "eleicao_prestacao_contas_candidato_origem_2022",
  "materialization_mode": "dev",
- "dbt_alias": False,
+ "dbt_alias": True,
  },
  ),
  ],
- filters=[filters.is_weekday],
 )

pipelines/datasets/br_jota/tasks.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+from prefect import task
+
+@task
+def get_table_ids() -> tuple[str, str, str, str]:
+
+ table_ids = ("eleicao_perfil_candidato_2024", "eleicao_prestacao_contas_candidato_origem_2024",
+ "eleicao_prestacao_contas_candidato_2024", "eleicao_prestacao_contas_partido_2024")
+
+ return table_ids