Add task to check for data updates and use it to

update metadata
basedosdados · Nov 6, 2023 · 662547f · 662547f
1 parent 46dfea3
commit 662547f
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 13 deletions.
diff --git a/pipelines/datasets/br_anatel_telefonia_movel/flows.py b/pipelines/datasets/br_anatel_telefonia_movel/flows.py
@@ -22,6 +22,7 @@
     clean_csv_municipio,
     clean_csv_uf,
     data_url,
+    task_check_for_data
 )
 from pipelines.utils.constants import constants as utils_constants
 from pipelines.utils.decorators import Flow
@@ -131,8 +132,7 @@
             )
 
         with case(update_metadata, True):
-            date = data_url()
-            date_string = str(date)  # task que retorna a data atual
+            date = task_check_for_data()
             update_django_metadata(
                 dataset_id,
                 table_id[0],
@@ -145,7 +145,7 @@
                 time_unit="months",
                 api_mode="prod",
                 date_format="yy-mm",
-                _last_date=date_string,
+                _last_date=date,
                 upstream_tasks=[wait_for_materialization],
             )
 
@@ -189,8 +189,7 @@
             )
 
         with case(update_metadata, True):
-            date = data_url()
-            date_string = str(date)  # task que retorna a data atual
+            date = task_check_for_data()
             update_django_metadata(
                 dataset_id,
                 table_id[1],
@@ -203,7 +202,7 @@
                 time_unit="months",
                 api_mode="prod",
                 date_format="yy-mm",
-                _last_date=date_string,
+                _last_date=date,
                 upstream_tasks=[wait_for_materialization],
             )
 
@@ -249,8 +248,7 @@
             )
 
         with case(update_metadata, True):
-            date = data_url()
-            date_string = str(date)  # task que retorna a data atual
+            date = task_check_for_data()
             update_django_metadata(
                 dataset_id,
                 table_id[2],
@@ -263,7 +261,7 @@
                 time_unit="months",
                 api_mode="prod",
                 date_format="yy-mm",
-                _last_date=date_string,
+                _last_date=date,
                 upstream_tasks=[wait_for_materialization],
             )
 
@@ -308,8 +306,7 @@
         )
 
         with case(update_metadata, True):
-            date = data_url()
-            date_string = str(date)  # task que retorna a data atual
+            date = task_check_for_data()
             update_django_metadata(
                 dataset_id,
                 table_id[3],
@@ -322,7 +319,7 @@
                 time_unit="months",
                 api_mode="prod",
                 date_format="yy-mm",
-                _last_date=date_string,
+                _last_date=date,
                 upstream_tasks=[wait_for_materialization],
             )
 

diff --git a/pipelines/datasets/br_anatel_telefonia_movel/tasks.py b/pipelines/datasets/br_anatel_telefonia_movel/tasks.py
@@ -4,7 +4,7 @@
 """
 
 import os
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 import numpy as np
 import pandas as pd
@@ -53,6 +53,7 @@ def data_url(anos, mes_um, mes_dois):
 @task
 def check_for_updates(dataset_id, table_id):
     data_obj = data_url("2023", "07", "12")
+    data_obj = datetime.strptime(data_obj, "%Y-%m-%d").date()
     # Obtém a última data no site BD
     data_bq_obj = extract_last_date(
         dataset_id, table_id, "yy-mm-dd", "basedosdados", data="data_coleta"
@@ -68,6 +69,9 @@ def check_for_updates(dataset_id, table_id):
     else:
         return False  # Não há novas atualizações disponíveis
 
+@task
+def task_check_for_data():
+    return data_url("2023", "07", "12")
 
 # ! TASK MICRODADOS
 @task(