prefeitura-rio · patriciacatandi · Sep 2, 2024 · Sep 3, 2024 · Sep 3, 2024 · Sep 3, 2024
diff --git a/.flake8 b/.flake8
@@ -1,2 +1,3 @@
 [flake8]
+ignore = E203,F401,F403, W503
 max-line-length = 100
diff --git a/pipelines/__init__.py b/pipelines/__init__.py
@@ -0,0 +1,6 @@
+# -*- coding: utf-8 -*-
+"""
+Imports all flows for every project so we can register all of them.
+"""
+from pipelines.meteorologia import *
+from pipelines.templates import *
diff --git a/pipelines/meteorologia/meteorologia_inmet/flows.py b/pipelines/meteorologia/meteorologia_inmet/flows.py
@@ -9,6 +9,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.meteorologia_inmet.schedules import hour_schedule
@@ -29,9 +30,7 @@
 
 with Flow(
     name="COR: Meteorologia - Meteorologia INMET",
-    code_owners=[
-        "paty",
-    ],
+    state_handlers=[handler_inject_bd_credentials],
 ) as cor_meteorologia_meteorologia_inmet:
     DATASET_ID = "clima_estacao_meteorologica"
     TABLE_ID = "meteorologia_inmet"

diff --git a/pipelines/meteorologia/meteorologia_inmet/tasks.py b/pipelines/meteorologia/meteorologia_inmet/tasks.py
@@ -11,10 +11,11 @@
 import pendulum
 import requests
 from prefect import task
+from prefeitura_rio.pipelines_utils.infisical import get_secret
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_alertario.utils import parse_date_columns
-from pipelines.utils.utils import get_vault_secret, log, to_partitions
+from pipelines.utils.utils import log, to_partitions
 
 # from pipelines.rj_cor.meteorologia.meteorologia_inmet.meteorologia_utils import converte_timezone
 
@@ -79,8 +80,7 @@ def download(data_inicio: str, data_fim: str) -> pd.DataFrame:
     # no UTC, visto que ele só traria dados do novo dia e substituiria
     # no arquivo da partição do dia atual no nosso timezone
 
-    dicionario = get_vault_secret("inmet_api")
-    token = dicionario["data"]["token"]
+    token = get_secret("INMET_API")["INMET_API"]
 
     raw = []
     for id_estacao in estacoes_unicas:

diff --git a/pipelines/meteorologia/meteorologia_redemet/flows.py b/pipelines/meteorologia/meteorologia_redemet/flows.py
@@ -9,6 +9,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.meteorologia_redemet.schedules import (
@@ -35,9 +36,7 @@
 
 with Flow(
     name="COR: Meteorologia - Meteorologia REDEMET",
-    code_owners=[
-        "paty",
-    ],
+    state_handlers=[handler_inject_bd_credentials],
 ) as cor_meteorologia_meteorologia_redemet:
     DUMP_MODE = Parameter("dump_mode", default="append", required=True)
     DATASET_ID = Parameter("dataset_id", default="clima_estacao_meteorologica", required=True)
@@ -49,7 +48,7 @@
 
     # Materialization parameters
     MATERIALIZE_AFTER_DUMP = Parameter("materialize_after_dump", default=True, required=False)
-    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=True, required=False)
+    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=False, required=False)
     MATERIALIZATION_MODE = Parameter("mode", default="prod", required=False)
 
     # Dump to GCS after? Should only dump to GCS if materializing to datario

diff --git a/pipelines/meteorologia/meteorologia_redemet/schedules.py b/pipelines/meteorologia/meteorologia_redemet/schedules.py
@@ -43,7 +43,7 @@
             parameter_defaults={
                 "materialize_after_dump": True,
                 "mode": "prod",
-                "materialize_to_datario": True,
+                "materialize_to_datario": False,
                 "dump_to_gcs": False,
                 # "dump_mode": "overwrite",
                 # "dataset_id": "clima_estacao_meteorologica",

diff --git a/pipelines/meteorologia/meteorologia_redemet/tasks.py b/pipelines/meteorologia/meteorologia_redemet/tasks.py
@@ -13,15 +13,11 @@
 from prefect import task
 from prefect.engine.signals import ENDRUN
 from prefect.engine.state import Failed
+from prefeitura_rio.pipelines_utils.infisical import get_secret
 from unidecode import unidecode
 
 from pipelines.constants import constants
-from pipelines.utils.utils import (
-    get_vault_secret,
-    log,
-    parse_date_columns,
-    to_partitions,
-)
+from pipelines.utils.utils import log, parse_date_columns, to_partitions
 
 
 @task(nout=3)
@@ -62,8 +58,7 @@ def download_data(first_date: str, last_date: str) -> pd.DataFrame:
         "SBSC",
     ]
 
-    redemet_token = get_vault_secret("redemet-token")
-    redemet_token = redemet_token["data"]["token"]
+    redemet_token = get_secret("REDEMET-TOKEN")["REDEMET-TOKEN"]
 
     # Converte datas em int para cálculo de faixas.
     first_date_int = int(first_date.replace("-", ""))
@@ -80,6 +75,7 @@ def download_data(first_date: str, last_date: str) -> pd.DataFrame:
                 res = requests.get(url)
                 if res.status_code != 200:
                     log(f"Problema no id: {id_estacao}, {res.status_code}")
+                    log(f"Data: {data}, Hora: {hora}")
                     continue
                 res_data = json.loads(res.text)
                 if res_data["status"] is not True:
@@ -90,12 +86,17 @@ def download_data(first_date: str, last_date: str) -> pd.DataFrame:
                     continue
                 raw.append(res_data)
 
+    log(f"Raw data size: {len(raw)}")
+
     # Extrai objetos de dataframe
     raw = [res_data["data"] for res_data in raw]
 
     # converte para dataframe
     dataframe = pd.DataFrame(raw)
 
+    # Log dataframe size
+    log(f"Daframe shape: {dataframe.shape}")
+
     return dataframe
 
 
@@ -205,8 +206,7 @@ def download_stations_data() -> pd.DataFrame:
     Download station information
     """
 
-    redemet_token = get_vault_secret("redemet-token")
-    redemet_token = redemet_token["data"]["token"]
+    redemet_token = get_secret("REDEMET-TOKEN")["REDEMET-TOKEN"]
     base_url = f"https://api-redemet.decea.mil.br/aerodromos/?api_key={redemet_token}"  # noqa
     url = f"{base_url}&pais=Brasil"
     res = requests.get(url)

diff --git a/pipelines/meteorologia/precipitacao_alertario/flows.py b/pipelines/meteorologia/precipitacao_alertario/flows.py
@@ -10,6 +10,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_alertario.constants import (
@@ -42,10 +43,7 @@
 
 with Flow(
     name="COR: Meteorologia - Precipitacao e Meteorologia ALERTARIO",
-    code_owners=[
-        "paty",
-    ],
-    # skip_if_running=True,
+    state_handlers=[handler_inject_bd_credentials],
 ) as cor_meteorologia_precipitacao_alertario:
     DATASET_ID_PLUVIOMETRIC = alertario_constants.DATASET_ID_PLUVIOMETRIC.value
     TABLE_ID_PLUVIOMETRIC = alertario_constants.TABLE_ID_PLUVIOMETRIC.value

diff --git a/pipelines/meteorologia/precipitacao_alertario/schedules.py b/pipelines/meteorologia/precipitacao_alertario/schedules.py
@@ -24,7 +24,7 @@
                 "materialize_after_dump_old_api": True,
                 "materialize_to_datario_old_api": True,
                 "materialize_after_dump": True,
-                "materialize_to_datario": True,
+                "materialize_to_datario": False,
                 "mode": "prod",
                 "dump_to_gcs": False,
             },

diff --git a/pipelines/meteorologia/precipitacao_alertario/tasks.py b/pipelines/meteorologia/precipitacao_alertario/tasks.py
@@ -13,6 +13,7 @@
 import requests
 from bs4 import BeautifulSoup
 from prefect import task
+from prefeitura_rio.pipelines_utils.infisical import get_secret
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_alertario.utils import (
@@ -23,7 +24,6 @@
     build_redis_key,
     compare_dates_between_tables_redis,
     get_redis_output,
-    get_vault_secret,
     log,
     parse_date_columns,
     save_str_on_redis,
@@ -42,8 +42,7 @@ def download_data() -> pd.DataFrame:
     Request data from API and return each data in a different dataframe.
     """
 
-    dicionario = get_vault_secret("alertario_api")
-    url = dicionario["data"]["url"]
+    url = get_secret("ALERTARIO_API")["ALERTARIO_API"]
 
     try:
         response = requests.get(url)

diff --git a/pipelines/meteorologia/precipitacao_cemaden/flows.py b/pipelines/meteorologia/precipitacao_cemaden/flows.py
@@ -9,6 +9,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_cemaden.constants import (
@@ -38,17 +39,15 @@
 
 with Flow(
     name="COR: Meteorologia - Precipitacao CEMADEN",
-    code_owners=[
-        "paty",
-    ],
+    state_handlers=[handler_inject_bd_credentials],
     # skip_if_running=True,
 ) as cor_meteorologia_precipitacao_cemaden:
     DUMP_MODE = Parameter("dump_mode", default="append", required=True)
     DATASET_ID = Parameter("dataset_id", default="clima_pluviometro", required=True)
     TABLE_ID = Parameter("table_id", default="taxa_precipitacao_cemaden", required=True)
     # Materialization parameters
     MATERIALIZE_AFTER_DUMP = Parameter("materialize_after_dump", default=True, required=False)
-    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=True, required=False)
+    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=False, required=False)
     MATERIALIZATION_MODE = Parameter("mode", default="prod", required=False)
     TRIGGER_RAIN_DASHBOARD_UPDATE = Parameter(
         "trigger_rain_dashboard_update", default=False, required=False

diff --git a/pipelines/meteorologia/precipitacao_cemaden/schedules.py b/pipelines/meteorologia/precipitacao_cemaden/schedules.py
@@ -23,7 +23,7 @@
                 # "trigger_rain_dashboard_update": True,
                 "materialize_after_dump": True,
                 "mode": "prod",
-                "materialize_to_datario": True,
+                "materialize_to_datario": False,
                 "dump_to_gcs": False,
                 "dump_mode": "append",
                 "dataset_id": "clima_pluviometro",

diff --git a/pipelines/meteorologia/precipitacao_inea/__init__.py b/pipelines/meteorologia/precipitacao_inea/__init__.py
diff --git a/pipelines/meteorologia/precipitacao_inea/flows.py b/pipelines/meteorologia/precipitacao_inea/flows.py
@@ -9,6 +9,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_inea.schedules import minute_schedule
@@ -34,9 +35,7 @@
 
 with Flow(
     name="COR: Meteorologia - Precipitacao e Fluviometria INEA",
-    code_owners=[
-        "paty",
-    ],
+    state_handlers=[handler_inject_bd_credentials],
     # skip_if_running=True,
 ) as cor_meteorologia_precipitacao_inea:
     DUMP_MODE = Parameter("dump_mode", default="append", required=True)
@@ -55,7 +54,7 @@
 
     # Materialization parameters
     MATERIALIZE_AFTER_DUMP = Parameter("materialize_after_dump", default=True, required=False)
-    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=True, required=False)
+    MATERIALIZE_TO_DATARIO = Parameter("materialize_to_datario", default=False, required=False)
     MATERIALIZATION_MODE = Parameter("mode", default="prod", required=False)
 
     # Dump to GCS after? Should only dump to GCS if materializing to datario

diff --git a/pipelines/meteorologia/precipitacao_inea/schedules.py b/pipelines/meteorologia/precipitacao_inea/schedules.py
@@ -23,7 +23,7 @@
                 # "trigger_rain_dashboard_update": True,
                 "materialize_after_dump": True,
                 "mode": "prod",
-                "materialize_to_datario": True,
+                "materialize_to_datario": False,
                 "dump_to_gcs": False,
                 "dump_mode": "append",
                 "dataset_id_pluviometric": "clima_pluviometro",

diff --git a/pipelines/meteorologia/precipitacao_websirene/flows.py b/pipelines/meteorologia/precipitacao_websirene/flows.py
@@ -7,6 +7,7 @@
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
 from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
+from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants
 from pipelines.meteorologia.precipitacao_websirene.schedules import MINUTE_SCHEDULE
@@ -24,9 +25,7 @@
 
 with Flow(
     "COR: Meteorologia - Precipitacao WEBSIRENE",
-    code_owners=[
-        "paty",
-    ],
+    state_handlers=[handler_inject_bd_credentials],
 ) as cor_meteorologia_precipitacao_websirene:
     DATASET_ID = "clima_pluviometro"
     TABLE_ID = "taxa_precipitacao_websirene"
@@ -85,6 +84,7 @@
 # para rodar na cloud
 cor_meteorologia_precipitacao_websirene.storage = GCS(constants.GCS_FLOWS_BUCKET.value)
 cor_meteorologia_precipitacao_websirene.run_config = KubernetesRun(
-    image=constants.DOCKER_IMAGE.value
+    image=constants.DOCKER_IMAGE.value,
+    labels=[constants.RJ_COR_AGENT_LABEL.value],
 )
 cor_meteorologia_precipitacao_websirene.schedule = MINUTE_SCHEDULE
diff --git a/pipelines/meteorologia/radar/mendanha/flows.py b/pipelines/meteorologia/radar/mendanha/flows.py
@@ -7,7 +7,7 @@
 from prefect import Parameter
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
-from prefeitura_rio.pipelines_utils.custom import Flow  # pylint: disable=E0611, E0401
+from prefeitura_rio.pipelines_utils.custom import Flow
 from prefeitura_rio.pipelines_utils.state_handlers import handler_inject_bd_credentials
 
 from pipelines.constants import constants

diff --git a/pipelines/meteorologia/satelite/flows.py b/pipelines/meteorologia/satelite/flows.py
@@ -9,6 +9,7 @@
 from prefect import Parameter, case
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
+from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
 from prefeitura_rio.pipelines_utils.custom import Flow  # pylint: disable=E0611, E0401
 from prefeitura_rio.pipelines_utils.state_handlers import (
     handler_initialize_sentry,

diff --git a/pipelines/rj_escritorio/__init__.py b/pipelines/rj_escritorio/__init__.py
diff --git a/pipelines/rj_escritorio/rain_dashboard/__init__.py b/pipelines/rj_escritorio/rain_dashboard/__init__.py