Call al SP

dags/components/Generation.py

@@ -144,8 +144,6 @@ def generate_and_deploy(command: str, intern_conn, params: Dict[str, Any], timez
             dataframe.to_csv(tmp_file, sep=delimiter, index=False, mode='a', header=header)
         except StopIteration:
             break
-
-
     list_outputs = params["s3_params"]
     size = os.path.getsize(tmp_file)
     for output in list_outputs:

dags/components/Transformation.py

@@ -98,9 +98,13 @@ def transformations(xcom_commands: str, intern_conn, timezone: str, **kwargs):
     with engine.connect() as connection:
         for command in commands:
             if any(command.startswith(palabra) or command.startswith(palabra.lower()) for palabra in not_procedure):
-                print(2)
-            logger.info(f"Ejecutando comando de transformación: {command}")
-            _ = connection.execute(command)
+                logger.info(f"Ejecutando comando de transformación: {command}")
+                _ = connection.execute(command)
+            else:
+                logger.info(f"Generando llamada al procedure según bd para: {command}")
+                command = intern_conn.generate_sql_procedure(command)
+                logger.debug(f"EJECUTANDO FINAL PROCEDURE: {command}")
+                _ = connection.execute(command)
     end_process_datetime = datetime_by_tzone(timezone).strftime('%d/%m/%Y %H:%M:%S')
     task.xcom_push(key="END_PROCESS_DATETIME_" + str(task.map_index), value=end_process_datetime)