DAG Reset process

dags/components/Control.py

@@ -48,7 +48,7 @@ def update_new_process(conf: List[Dict[str, Any]], status: str, tasks: Dict[str,
             tasks.update(last_tasks)
             conf.pop(-1)
         current_datetime = str(datetime_by_tzone(timezone, '%Y-%m-%d %H:%M:%S'))
-        new_process = {"date": current_datetime, "status": status, "tasks": tasks, "created_tables": created_tables}
+        new_process = {"date": current_datetime, "status": status, "tasks": tasks, "objects_created": created_tables}
         if current_period == processed_period and isinstance(conf, List):
             conf.append(new_process)
         else:

dags/components/Generation.py

@@ -140,6 +140,7 @@ def generate_and_deploy(command: str, intern_conn, params: Dict[str, Any], timez
                         dataframe[campo] = pd.to_datetime(dataframe[campo]).dt.date
                     # elif campos[campo] == DataTypeEnum.DATETIME.name:  # datetime:
                     #     dataframe[campo] = pd.to_datetime(dataframe[campo], format='%Y-%m-%d %H:%M:%S')
+
             dataframe = dataframe.drop("INTERN_ID_BCOM", axis=1, errors='ignore')
             logger.debug(dataframe)
             dataframe.to_csv(tmp_file, sep=delimiter, index=False, mode='a', header=header)

dags/dag_conf.yml

@@ -55,15 +55,15 @@ app:
     delimiter: '|'
     tmp_path: /tmp
     s3_params:
-        ESTUDIANTES_1:
+        tabla4:
           bucket: pruebairflow
           prefix: bcom_results
           connection_id: prueba_af
-        ESTUDIANTES_11:
+        tabla1:
           bucket: pruebairflow
           prefix: bcom_results
           connection_id: prueba_af
-        CATALOGO_PROMOCIONES:
+        tabla2:
           bucket: pruebairflow
           prefix: bcom_results
           connection_id: prueba_af
@@ -74,7 +74,7 @@ app:
       connection_id: conn_script
       filename: report_<datetime>.xlsx
     datetime_pattern: '%Y-%m-%d %H:%M:%S'
-  procedure:
+  definitions:
     filepath: "/opt/airflow/dags/procedure_prueba.json"
 
 

dags/dag_reset_process.py

@@ -112,7 +112,7 @@ def set_dag():
         control_s3 = conf["control"]["s3_params"]
         timezone = conf["timezone"]
         # NUEVA VARIABLE
-        procedure = conf["procedure"]
+        procedure = conf["definitions"]
         control_extractor = PythonOperator(
             task_id="CONTROL-EXTRACTOR",
             python_callable=extract_last_control,

dags/dag_transformacion_bcom.py

@@ -138,7 +138,6 @@ def set_dag():
     conf_path = MAIN_PATH + "dag_conf.yml"
     with open(conf_path) as f:
         data = yaml.load(f, Loader=SafeLoader)
-        logger.info(f"CONFIGURACIÓN: {data}")
         conf = data["app"]
     with DAG(DAG_NAME, default_args=DEFAULT_ARGS, description="Proceso que extrae y transforma",
              schedule_interval=conf["schedule"], tags=["DAG BCOM - SQL TRANSFORMATIONS"], catchup=False) as dag:
@@ -152,7 +151,7 @@ def set_dag():
         control_s3 = conf["control"]["s3_params"]
         timezone = conf["timezone"]
         # NUEVA VARIABLE
-        procedure = conf["procedure"]
+        procedure = conf["definitions"]
         # Scripts extraction
         extract_mask = conf["source_mask"]
         transform_mask = conf["transformation_mask"]

dags/procedure_prueba.json

 [
   {
-    "identifier": "obtenerEstudiantes",
+    "identifier": "ModificarTabla4",
     "transformation_store_procedure": true
   },
   {
-    "identifier": "TempEstudiantes",
+    "identifier" : "UnionYInsert",
+    "transformation_store_procedure": true
+  },
+  {
+    "identifier": "TempTabla",
     "temp_table" : true
   },
   {
-    "identifier": "ESTUDIANTES_11",
+    "identifier": "tabla1",
     "fields": [
       {
-        "name": "ID",
+        "name": "id",
         "datatype": "NUMBER",
         "decimal_precision": 0
       },
       {
-        "name": "Nombre",
+        "name": "nombre",
         "datatype": "TEXT",
         "maxLength": 50
       },
       {
-        "name": "Apellido",
+        "name": "fecha_nacimiento",
+        "datatype": "DATE"
+      }
+    ],
+    "indexes": [
+      {
+        "name" : "indice1",
+        "index_fields": [
+          "id"
+        ]
+      }
+    ],
+    "save_output" : false
+  },
+  {
+    "identifier": "tabla2",
+    "fields": [
+      {
+        "name": "id",
+        "datatype": "NUMBER",
+        "decimal_precision": 0
+      },
+      {
+        "name": "apellido",
         "datatype": "TEXT",
         "maxLength": 50
       },
       {
-        "name": "Edad",
+        "name": "fecha_registro",
+        "datatype": "DATETIME"
+      }
+    ],
+    "indexes": [
+      {
+        "name" : "indice1",
+        "index_fields": [
+          "id"
+        ]
+      }
+    ],
+    "save_output" : false
+  },
+  {
+    "identifier": "tabla3",
+    "fields": [
+      {
+        "name": "id",
         "datatype": "NUMBER",
         "decimal_precision": 0
       },
       {
-        "name": "CorreoElectronico",
-        "datatype": "TEXT",
-        "maxLength": 100
+        "name": "edad",
+        "datatype": "NUMBER",
+        "decimal_precision": 0
+      },
+      {
+        "name": "fecha_actualizacion",
+        "datatype": "DATETIME"
       }
     ],
     "indexes": [
       {
         "name" : "indice1",
         "index_fields": [
-          "ID"
+          "id"
         ]
       }
     ],
-    "save_output" : true
+    "save_output" : false
   },
   {
-    "identifier": "ESTUDIANTES_1",
+    "identifier": "tabla4",
     "fields": [
       {
         "name": "id",
         "datatype": "NUMBER",
         "decimal_precision": 0
       },
+      {
+        "name": "nombre",
+        "datatype": "TEXT",
+        "maxLength": 50
+      },
       {
         "name": "fecha",
         "datatype": "DATE"
       },
       {
-        "name": "fecha_tiempo",
+        "name": "fecha_2",
+        "datatype": "DATETIME"
+      },
+      {
+        "name": "fecha_3",
         "datatype": "DATETIME"
       }
     ],