Optimization of extraction process from an Oracle DB

dags/components/DatabaseOperation/DatabaseLoad.py

@@ -11,25 +11,14 @@ def save_from_dataframe(df: pd.DataFrame, tablename: str, connection) -> bool:
     save = False
     try:
         chunksize = 2000
-        # db_type = connection.db_type
+        db_type = connection.db_type
         connection = connection.engine
-        # print(df["CREACION_PRODUCTO"].value_counts())
         with connection.connect() as conn:
-            # if db_type == DatabaseTypeEnum.ORACLE.value:
-            #     df.info()
-            #     aux = df.columns[df.dtypes == 'object'].tolist()
-            #     print(aux)
-            #     dtyp = {}
-            #     for col in aux:
-            #         print(col)
-            #         print(df[col].dtype)
-            #         df[col] = df[col].astype(str)
-            #         dtyp.update({col: VARCHAR(df[col].str.len().max())})
-            #     # dtyp = {c: VARCHAR(df[c].str.len().max()) for c in aux}
-            #     print(dtyp)
-            #     df.to_sql(tablename, conn, if_exists='append', dtype=dtyp, index=False, chunksize=chunksize)
-            # else:
-            df.to_sql(tablename, conn, if_exists='append', index=False, chunksize=chunksize)
+            if db_type == DatabaseTypeEnum.ORACLE.value:
+                dtypes = {c: VARCHAR(df[c].str.len().max()) for c in df.columns[df.dtypes == 'object'].tolist()}
+                df.to_sql(tablename, conn, if_exists='append', dtype=dtypes, index=False, chunksize=chunksize)
+            else:
+                df.to_sql(tablename, conn, if_exists='append', index=False, chunksize=chunksize)
             save = True
     except Exception as e:
         logger.error(f"Error guardando resultados desde dataframe. {e}")

dags/components/S3Route.py

@@ -115,7 +115,6 @@ def save_df_to_s3(data: pd.DataFrame or str, conn: str, bucket: str, key: str, p
                 else:
                     hook.load_bytes(buffer.getvalue(), key, bucket, True)
             else:
-                logger.info(f"Llegue aca, para guardar")
                 logger.info(f"DATA: {data}. BUCKET: {bucket}. KEY: {key}")
                 # convert_df = pd.read_csv()
                 if gcp_cloud:

dags/components/Transformation.py

@@ -8,12 +8,12 @@ from airflow.exceptions import AirflowSkipException
 from components.DatabaseOperation.DatabaseTransformation import delete_table, delete_procedure
 from components.S3Route import get_files_from_prefix, get_file_from_prefix
 from components.Xcom import save_commands_to_xcom
-from components.Utils import update_sql_commands_2
 from components.Control import get_tasks_from_control, update_new_process
 from components.S3Route import load_control_to_s3
 from components.Xcom import delete_all_xcom_tasks, delete_task_instances
 from enums.ProcessStatusEnum import ProcessStatusEnum
 from enums.OperationTypeEnum import OperationTypeEnum
+from enums.DatabaseTypeEnum import DatabaseTypeEnum
 from components.Timezone import datetime_by_tzone
 
 import logging
@@ -99,12 +99,16 @@ def transformations(xcom_commands: str, intern_conn, timezone: str, **kwargs):
     script_name = xcom_commands[0]
     commands = xcom_commands[1]
     logger.info(f"Ejecutando transformaciones del script {script_name}")
-    not_procedure = ["UPDATE","SELECT","CREATE","ALTER","DROP","DELETE","INSERT","GRANT","REVOKE","TRUNCATE","COPY",
-                     "COMMIT","ROLLBACK","USE","BEGIN"]
+    logger.info(f"XCOM_COMMANDS: {xcom_commands}")
+    not_procedure = ["UPDATE", "SELECT", "CREATE", "ALTER", "DROP", "DELETE", "INSERT"," GRANT", "REVOKE", "TRUNCATE",
+                     "COPY", "COMMIT", "ROLLBACK", "USE", "BEGIN"]
     with engine.connect() as connection:
         for command in commands:
+            logger.info(f"Commando: {command}")
             if any(command.startswith(palabra) or command.startswith(palabra.lower()) for palabra in not_procedure):
-                logger.info(f"Ejecutando comando de transformación: {command}")
+                logger.info(f"Ejecutando comando de transformación en {intern_conn.db_type} : {command}")
+                if intern_conn.db_type == DatabaseTypeEnum.ORACLE.value:
+                    command = command[:-1]
                 _ = connection.execute(command)
             else:
                 logger.info(f"Generando llamada al procedure según bd para: {command}")
@@ -150,7 +154,9 @@ def get_trans_from_xcom(provider, store_procedure, procedure_mask, label_tablena
                         # result = db_intern_conn.check_procedure(tablename, connection)
                         # if not result:
                         try:
+                            logger.info("1")
                             _ = connection.execute(procedure)
+                            logger.info("2")
                         except Exception as e:
                             logger.error(f" Error: {e}")
                             raise AirflowSkipException
@@ -159,7 +165,7 @@ def get_trans_from_xcom(provider, store_procedure, procedure_mask, label_tablena
             else:
                 logger.debug(f"No se encontró el label en {procedure} por ende no se creará")
     save_commands_to_xcom(procedures, kwargs['ti'], procedure_mask, transform_mask, procedure_mask, order_delimiter)
-    logger.debug(f"Procedures cargados en Xcom: {procedures}")
+    logger.info(f"Procedures cargados en Xcom: {procedures}")
 
     transforms_per_file = []
     conf = task.xcom_pull(task_ids="VALIDATE_EXTRACTION", key="CONTROL-CONFIG")