Update 30-07-23. Fix some bugs about Postgres and Oracle DB

dags/components/Databases/Database.py

@@ -55,10 +55,11 @@ class Database:
         except Exception as e:
             logger.error(f"Error creando db engine. {e}")
 
-    def create_model(self, tablename: str, fields: List[Tuple[str]], modelName: str = "TableModel") -> bool:
+    def create_model(self, tablename: str, fields: List[Tuple[str]], db_target: str,
+                     modelName: str = "TableModel") -> bool:
         create = False
         try:
-            create = self.factory.create_model(tablename, fields, modelName)
+            create = self.factory.create_model(tablename, fields, db_target, modelName)
         except Exception as e:
             raise AssertionError(f"Error creando tabla dinámica con nombre {tablename}. {e}")
         finally:

dags/components/Databases/Mysql.py

@@ -3,7 +3,8 @@ from typing import List, Tuple
 import pymysql
 from sqlalchemy import create_engine
 from enums.DatabaseDialectEnum import DatabaseDialectEnum
-from components.Model.InsumoModel import InsumoModel
+from enums.DatabaseTypeEnum import DatabaseTypeEnum
+from components.Model.InsumoModel import InsumoModel, InsumoModelOracle
 from components.Databases.Enums.MysqlDataTypeEnum import MysqlDataTypeEnum
 from components.Databases.Enums.MysqlDataTypeORMEnum import MysqlDataTypeORMEnum
 
@@ -43,13 +44,17 @@ class Mysql:
         except Exception as e:
             logger.error(f"Error creando engine de Mysql. {e}")
 
-    def create_model(self, tablename: str, fields: List[Tuple[str]], modelName: str = "TableModel") -> bool:
+    def create_model(self, tablename: str, fields: List[Tuple[str]], db_target: str, modelName: str = "TableModel"):
         model = None
         try:
-            model = type(modelName, (InsumoModel,), {
+            db_args = {
                 '__tablename__': tablename,
                 '__table_args__': {'extend_existing': True}
-            })
+            }
+            if db_target == DatabaseTypeEnum.ORACLE.value:
+                model = type(modelName, (InsumoModelOracle,), db_args)
+            else:
+                model = type(modelName, (InsumoModel,), db_args)
             for field in fields:
                 logger.debug(f"Attribute: {field}")
                 name = field[0]

dags/components/Databases/Oracle.py

@@ -4,9 +4,10 @@ from sqlalchemy import create_engine
 import oracledb
 
 from enums.DatabaseDialectEnum import DatabaseDialectEnum
+from enums.DatabaseTypeEnum import DatabaseTypeEnum
 from components.Databases.Enums.OracleDataTypeEnum import OracleDataTypeEnum
 from components.Databases.Enums.OracleDataTypeORMEnum import OracleDataTypeORMEnum
-from components.Model.InsumoModel import InsumoModel
+from components.Model.InsumoModel import InsumoModel, InsumoModelOracle
 from sqlalchemy import Column
 import logging
 
@@ -43,13 +44,17 @@ class Oracle:
         except Exception as e:
             logger.error(f"Error creando engine de Oracle. {e}")
 
-    def create_model(self, tablename: str, fields: List[Tuple[str]], modelName: str = "TableModel") -> bool:
+    def create_model(self, tablename: str, fields: List[Tuple[str]], db_target: str, modelName: str = "TableModel"):
         model = None
         try:
-            model = type(modelName, (InsumoModel,), {
+            db_args = {
                 '__tablename__': tablename,
                 '__table_args__': {'extend_existing': True}
-            })
+            }
+            if db_target == DatabaseTypeEnum.ORACLE.value:
+                model = type(modelName, (InsumoModelOracle,), db_args)
+            else:
+                model = type(modelName, (InsumoModel,), db_args)
             for field in fields:
                 logger.debug(f"Attribute: {field}")
                 name = field[0]

dags/components/Databases/Postgres.py

@@ -2,8 +2,9 @@ from typing import List, Tuple
 
 import psycopg2
 from sqlalchemy import create_engine
-from components.Model.InsumoModel import InsumoModel
+from components.Model.InsumoModel import InsumoModel, InsumoModelOracle
 from enums.DatabaseDialectEnum import DatabaseDialectEnum
+from enums.DatabaseTypeEnum import DatabaseTypeEnum
 from components.Databases.Enums.PostgresDataTypeEnum import PostgresDataTypeEnum
 from components.Databases.Enums.PostgresDataTypeORMEnum import PostgresDataTypeORMEnum
 
@@ -46,13 +47,17 @@ class Postgres:
         except Exception as e:
             logger.error(f"Error creando engine de Postgres. {e}")
 
-    def create_model(self, tablename: str, fields: List[Tuple[str]], modelName: str = "TableModel"):
+    def create_model(self, tablename: str, fields: List[Tuple[str]], db_target: str, modelName: str = "TableModel"):
         model = None
         try:
-            model = type(modelName, (InsumoModel,), {
+            db_args = {
                 '__tablename__': tablename,
                 '__table_args__': {'extend_existing': True}
-            })
+            }
+            if db_target == DatabaseTypeEnum.ORACLE.value:
+                model = type(modelName, (InsumoModelOracle,), db_args)
+            else:
+                model = type(modelName, (InsumoModel,), db_args)
             for field in fields:
                 logger.debug(f"Attribute: {field}")
                 name = field[0]

dags/components/Extractor.py

 from typing import Any, Dict
 import json
 
+import numpy as np
 import pandas as pd
 from enums.DatabaseTypeEnum import DatabaseTypeEnum
 from enums.ProcessStatusEnum import ProcessStatusEnum
@@ -113,6 +114,9 @@ def extract_from_source(command: str, source_conn, intern_conn, chunksize: int,
     else:
         if source_conn.db_type == DatabaseTypeEnum.ORACLE.value and \
                 not extract_type.startswith(OperationTypeEnum.PROCEDURE.value):
+            values = command_for_create.split("|")
+            if len(values) > 1:
+                command_for_create = values[1]
             command_for_create = f"SELECT * FROM ({command_for_create}) OFFSET 1 ROWS FETCH NEXT 1 ROWS ONLY"
         else:
             command_words = command_for_create.split(" ")
@@ -131,7 +135,7 @@ def extract_from_source(command: str, source_conn, intern_conn, chunksize: int,
         logger.debug(f"Columnas procesadas: {columns}")
         multiple = select_multiple(command)
         tablename = multiple["tablename"]
-        model = source_conn.create_model(tablename, columns)
+        model = source_conn.create_model(tablename, columns, intern_conn.db_type)
     try:
         create = intern_conn.create_table(model)
         if create:
@@ -191,8 +195,7 @@ def extract_from_source(command: str, source_conn, intern_conn, chunksize: int,
             logger.info(f"Número de pasos para migrar datos: {steps}")
             for step in range(steps):
                 dataframe = next(iterator)
-                dataframe["INTERN_ID_BCOM"] = None
-                logger.debug(dataframe)
+                dataframe = dataframe.fillna(value=np.nan)
                 save = save_from_dataframe(dataframe, tablename, intern_conn.engine)
                 if save:
                     logger.info(f"Guardado correctamente dataframe en el paso {step+1}")

dags/components/Model/InsumoModel.py

 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import Column, BigInteger
+from sqlalchemy.schema import Identity
 
 Base = declarative_base()
 
@@ -9,3 +10,10 @@ class InsumoModel(Base):
     __abstract__ = True
 
     INTERN_ID_BCOM = Column(BigInteger, primary_key=True, autoincrement=True)
+
+
+class InsumoModelOracle(Base):
+
+    __abstract__ = True
+
+    INTERN_ID_BCOM = Column(BigInteger, Identity(start=1), primary_key=True)