Fix bugs deleting tables

dags/components/Cleaning.py

@@ -29,7 +29,7 @@ def validate_clean(control_params: Dict[str, Any], provider: str, timezone: str,
                 result_p = bool(intern_conn.check_procedure(procedure, conn))
                 if result_p:
                     logger.info(f"Borrando procedures {procedure}")
-                    delete = delete_procedure(procedure, engine)
+                    delete = delete_procedure(procedure, intern_conn)
                     if delete:
                         logger.info(f"Borrado correctamente el procedure {procedure}")
     success_tasks = ti.xcom_pull(task_ids="GENERATORS", key="SUCCESS_TASKS")
@@ -41,7 +41,6 @@ def validate_clean(control_params: Dict[str, Any], provider: str, timezone: str,
     if not prefix.endswith("/"):
         prefix += "/"
     key = prefix + control_params["filename"]
-    #conf = json.dumps(conf, indent=2, default=str)
     final_dict = {}
     status = ProcessStatusEnum.SUCCESS.value
     if not isinstance(success_tasks, type(None)) and len(success_tasks) > 0:
@@ -62,19 +61,9 @@ def validate_clean(control_params: Dict[str, Any], provider: str, timezone: str,
 
 
 def clean(command: str, intern_conn):
-    engine = intern_conn.engine
     tablename = command
-    # with engine.connect() as conn:
-    #     result_t = bool(intern_conn.check_table(tablename, conn))
-    #     result_p = bool(intern_conn.check_procedure(tablename, conn))
-    # if result_t:
-    #     logger.info(f"Borrando tabla {tablename}")
-    #     delete_t = delete_table(tablename, engine)
-    # elif result_p:
-    #     logger.info(f"Borrando procedure {tablename}")
-    #     delete_p = delete_procedure(tablename, engine)
-    delete_p = delete_procedure(tablename, engine)
-    delete_t = delete_table(tablename, engine)
+    delete_p = delete_procedure(tablename, intern_conn)
+    delete_t = delete_table(tablename, intern_conn)
     if delete_t:
         logger.info(f"Borrado correctamente la tabla {tablename}")
     elif delete_p:

dags/components/DatabaseOperation/DatabaseLoad.py

@@ -7,7 +7,7 @@ import logging
 logger = logging.getLogger()
 
 
-def save_from_dataframe(df: pd.DataFrame, tablename: str, connection) -> bool:
+def save_from_dataframe(df: pd.DataFrame, tablename: str, connection, is_first: bool = False) -> bool:
     save = False
     try:
         chunksize = 2000

dags/components/DatabaseOperation/DatabaseTransformation.py

@@ -2,6 +2,9 @@ import logging
 import time
 from typing import List
 import re
+
+from enums.DatabaseTypeEnum import DatabaseTypeEnum
+
 logger = logging.getLogger()
 
 
@@ -15,11 +18,11 @@ def execute_transformations(commands: List[str], engine):
         logger.error(f"Error ejecutando comando de transformación. {e}")
 
 
-def delete_table(tablename: str, engine) -> bool:
-    base_Datos = engine.dialect.name
+def delete_table(tablename: str, connection) -> bool:
+    engine = connection.engine
     delete = False
     try:
-        if base_Datos == "oracle":
+        if connection.db_type == DatabaseTypeEnum.ORACLE.value:
             command = f'DROP TABLE {tablename}'
             start_time = time.time()
             with engine.connect() as conn:
@@ -33,7 +36,7 @@ def delete_table(tablename: str, engine) -> bool:
                     _ = conn.execute(command)
                     delete = True
                     logger.debug(f"Duración de borrado: {time.time() - start_time}")
-        elif base_Datos == "mysql":
+        else:
             command = f'DROP TABLE IF EXISTS {tablename}'
             start_time = time.time()
             with engine.connect() as conn:
@@ -49,11 +52,11 @@ def delete_table(tablename: str, engine) -> bool:
         return delete
 
 
-def delete_procedure(procedure: str, engine) -> bool:
-    base_Datos = engine.dialect.name
+def delete_procedure(procedure: str, connection) -> bool:
+    engine = connection.engine
     delete = False
     try:
-        if base_Datos == "oracle":
+        if connection.db_type == DatabaseTypeEnum.ORACLE.value:
             command = f"DROP PROCEDURE {procedure}"
             start_time = time.time()
             with engine.connect() as conn:
@@ -68,7 +71,7 @@ def delete_procedure(procedure: str, engine) -> bool:
                     _ = conn.execute(command)
                     delete = True
                     logger.debug(f"Duración de borrado: {time.time() - start_time}")
-        elif base_Datos == "mysql":
+        else:
             command = f'DROP PROCEDURE IF EXISTS {procedure}'
             start_time = time.time()
             with engine.connect() as conn:

dags/components/Databases/Mysql.py

@@ -135,4 +135,4 @@ class Mysql:
         except Exception as e:
             logger.error(f"Error obteniendo counts de tabla. {e}")
         finally:
-            return exists
+            return exists
\ No newline at end of file

dags/components/Databases/Oracle.py

@@ -90,7 +90,6 @@ class Oracle:
     def generate_sql_procedure(self, command: str, reserved_word: str = "execute") -> str:
         response = ""
         try:
-            logger.info("COMANDO",command)
             command = command.replace(reserved_word, "").replace(";", "")
             response = f"BEGIN {command}; END;"
             logger.debug("COMANDO ORACLE:", response)

dags/components/Extractor.py

@@ -145,8 +145,6 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez
             create = intern_conn.create_table(model)
             if create:
                 logger.info(f"Creado correctamente la tabla {tablename}. Creado?: {create}")
-                #extract_tables.append(tablename)
-                #task.xcom_push(key="TABLES_CREATED", value=tablename)
             else:
                 raise AssertionError(f"Error creando tabla {tablename}")
             if is_procedure:
@@ -225,7 +223,7 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez
         task.xcom_push(key="TABLES_CREATED", value=tablename)
         task.xcom_push(key="END_PROCESS_DATETIME_" + str(task.map_index), value=end_process_datetime)
     except Exception as e:
-        delete = delete_table(tablename, intern_conn.engine)
+        delete = delete_table(tablename, intern_conn)
         if delete:
             logger.info(f"Se borró correctamente la tabla {tablename}")
         raise AssertionError(f"Error creando la tabla y migrando datos. {type(e)}. {e}")
@@ -240,7 +238,6 @@ def get_select_from_xcom(db_intern_conn, **kwargs):
     conf = task.xcom_pull(task_ids="SCRIPTS-EXTRACTOR", key="CONTROL-CONFIG")
     definitions = task.xcom_pull(task_ids="SCRIPTS-EXTRACTOR", key="EXTRACTION-DEFINITION-JSON")
     temp_conf = conf[-1]["objects_created"] if "objects_created" in conf[-1].keys() else []
-    engine = db_intern_conn.engine
     logger.info(f"OBJETOS CREADOS: {temp_conf}")
     identificadores = [item["identifier"] for item in definitions]
     if temp_conf:
@@ -258,8 +255,8 @@ def get_select_from_xcom(db_intern_conn, **kwargs):
         tablas_temp = list(tablas_temp)
         if tablas_temp:
             for i in tablas_temp:
-                delete = delete_table(i, engine)
-                delete2 = delete_procedure(i, engine)
+                delete = delete_table(i, db_intern_conn)
+                delete2 = delete_procedure(i, db_intern_conn)
                 if delete:
                     logger.info(f"Borrado correctamente la tabla temporal {i}")
                 if delete2:

dags/components/Transformation.py

@@ -147,12 +147,10 @@ def get_trans_from_xcom(provider, store_procedure, procedure_mask, label_tablena
                 index = procedure.find(label_tablename + ":")
                 label_lenght = len(label_tablename + ":")
                 tablename = procedure[index + label_lenght:].strip().split("\n")[0]
-                delete_procedure(tablename, engine)
+                delete_procedure(tablename, db_intern_conn)
                 for element in definitions:
                     if element["identifier"] == tablename and "transformation_store_procedure" in element.keys() and element["transformation_store_procedure"] == True:
                         logger.info(f"Ejecutando creacion de procedure: {procedure}")
-                        # result = db_intern_conn.check_procedure(tablename, connection)
-                        # if not result:
                         try:
                             logger.info("1")
                             _ = connection.execute(procedure)

dags/components/Utils.py

@@ -166,13 +166,6 @@ def update_sql_commands(dataset: List[Tuple[str, str]], label_tablename: str,
             for item in data:
                 if item.lower().strip() == "end":
                     final_data[-1] = final_data[-1] + "; end;"
-                # parts = item.split(CommentsScriptEnum.DASHES.value)
-                # parts = [part for part in parts if len(part.strip()) > 0]
-                # print(parts)
-                # if len(parts) > 1:
-                #     for part in parts:
-                #         if not part.strip().lower().startswith(label_tablename.lower()):
-                #             continue
 
                 final_item = item
                 if item.lower().strip().find(label_tablename.lower().strip() + ":") != -1:
@@ -222,14 +215,6 @@ def select_multiple(command: str) -> Dict[str, Any]:
             if command.lower().replace(" ", "").find(no_procedure_init) != -1:
                 response["is_multiple"] = True
                 tablename = command[:command.index("|")].strip()
-            # init_index = command.lower().find("from")
-            # if init_index == -1:
-            #     raise AssertionError("Query malformed")
-            # else:
-            #     from_command = command[init_index + 4:]
-            #     tablename_base = from_command.strip().split(" ")
-            #     if len(tablename_base) > 0 and tablename == "":
-            #         tablename = tablename_base[0]
             response["tablename"] = tablename
     except Exception as e:
         raise AssertionError(f"Error validando si es múltiple select y nombre de tabla. {e}")