Skip to content

B
bcom-tp-etl-transformation-pipelines
Commit ad679d5e authored by Erly Villaroel's avatar Erly Villaroel
Browse files
Options

Cambios en Oracle.py

parent 8ebc05c1
Hide whitespace changes
Inline Side-by-side
Showing with 44 additions and 150 deletions
dags/components/DatabaseOperation/DatabaseTransformation.py
View file @ ad679d5e
......@@ -18,7 +18,7 @@ def execute_transformations(commands: List[str], engine):
def delete_table(tablename: str, engine) -> bool:
delete = False
try:
command = f'DROP TABLE IF EXISTS {tablename}'
command = f'DROP TABLE {tablename}'
start_time = time.time()
with engine.connect() as conn:
try:
......@@ -36,7 +36,7 @@ def delete_table(tablename: str, engine) -> bool:
def delete_procedure(procedure: str, engine) -> bool:
delete = False
try:
command = f"DROP PROCEDURE IF EXISTS {procedure}"
command = f"DROP PROCEDURE {procedure}"
start_time = time.time()
with engine.connect() as conn:
try:
......
dags/components/Databases/Oracle.py
View file @ ad679d5e
......@@ -91,7 +91,7 @@ class Oracle:
response = ""
try:
command = command.replace(reserved_word, "").replace(";", "")
response = f"begin {command}; end;"
response = f"BEGIN {command}; END;"
logger.debug("COMANDO ORACLE:", response)
except Exception as e:
logger.error(f"Error generando comando sql para procedure Oracle. Comando: {command}. {e}")
......@@ -113,7 +113,7 @@ class Oracle:
def check_procedure(self, procedure_name, connection) -> bool:
exists = False
try:
check_query = f"SELECT text FROM all_source WHERE name = '{procedure_name}'"
check_query = f"SELECT object_name FROM all_procedures WHERE object_name = '{procedure_name}'"
result = connection.execute(check_query)
exists = bool(result.fetchone())
except Exception as e:
......@@ -124,7 +124,7 @@ class Oracle:
def check_table(self, table_name, connection) -> bool:
exists = False
try:
check_query = f"DESCRIBE {table_name}"
check_query = f"SELECT table_name FROM all_tables WHERE table_name = '{table_name}'"
result = connection.execute(check_query)
exists = bool(result.fetchone())
except Exception as e:
......
dags/dag_conf.yml
View file @ ad679d5e
......@@ -5,23 +5,23 @@ app:
database:
sources:
source1:
type: mysql
host: 192.168.1.13
port: 13306
username: root
password: root
database: prueba
service:
schema: sources
type: oracle
host: 192.168.27.22
port: 21521
username: PRUEBABCOM2
password: admin
database:
service: ORCLPDB1
schema:
transformation:
type: mysql
host: 192.168.1.13
port: 13306
username: root
password: root
database: prueba_ca
service:
schema: intern_db
type: oracle
host: 192.168.27.22
port: 21521
username: RLQA_AIR
password: RLQA_AIR99
database:
service: ORCLPDB1
schema:
chunksize: 4000
label_multiple_select: TABLENAME
label_transform_procedure: STORE
......@@ -55,14 +55,14 @@ app:
delimiter: '|'
tmp_path: /tmp
s3_params:
tabla4:
bucket: pruebairflow
tabla1:
bucket: prueba-id
prefix: bcom_results
connection_id: prueba_af
connection_id: conn_script
tabla5:
bucket: pruebairflow
bucket: prueba-id
prefix: bcom_results
connection_id: prueba_af
connection_id: conn_script
report:
s3_params:
bucket: prueba-id
......
dags/procedure_prueba.json
View file @ ad679d5e
[
{
"identifier": "ModificarTabla4",
"identifier": "SP_ACTUALIZAR_TABLA_5",
"transformation_store_procedure": true
},
{
"identifier" : "UnionYInsert",
"transformation_store_procedure": true
},
{
"identifier": "TempTabla",
"temp_table" : true
},
{
"identifier": "tabla11",
"identifier": "tabla1",
"fields": [
{
"name": "id",
"datatype": "NUMBER",
"decimal_precision": 0
},
{
"name": "nombre",
"name": "CD_FOLIO",
"datatype": "TEXT",
"maxLength": 50
"maxLength": 100
},
{
"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",
"name": "CD_CUENTA",
"datatype": "TEXT",
"maxLength": 50
},
{
"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": "edad",
"datatype": "NUMBER",
"decimal_precision": 0
"maxLength": 100
},
{
"name": "fecha_actualizacion",
"datatype": "DATETIME"
}
],
"indexes": [
{
"name" : "indice1",
"index_fields": [
"id"
]
}
],
"save_output" : false
},
{
"identifier": "tabla4",
"fields": [
{
"name": "id",
"datatype": "NUMBER",
"decimal_precision": 0
},
{
"name": "nombre",
"name": "CD_PAQUETE",
"datatype": "TEXT",
"maxLength": 50
"maxLength": 100
},
{
"name": "fecha",
"datatype": "DATE"
},
{
"name": "fecha_2",
"datatype": "DATE"
},
{
"name": "fecha_3",
"datatype": "DATETIME"
"name": "NB_PAQUETE",
"datatype": "TEXT",
"maxLength": 200
}
],
"indexes": [
{
"name" : "indice1",
"index_fields": [
"id"
"CD_FOLIO"
]
}
],
......@@ -135,36 +41,24 @@
"identifier": "tabla5",
"fields": [
{
"name": "id",
"datatype": "NUMBER",
"decimal_precision": 0
"name": "NOMBRE_PRODUCTO",
"datatype": "TEXT",
"maxLength": 100
},
{
"name": "nombre",
"name": "CD_PAQUETE",
"datatype": "TEXT",
"maxLength": 50
},
{
"name": "fecha",
"datatype": "DATE"
},
{
"name": "fecha_2",
"datatype": "DATETIME"
},
{
"name": "fecha_3",
"datatype": "DATETIME"
}
],
"indexes": [
{
"name" : "indice1",
"index_fields": [
"id"
"NOMBRE_PRODUCTO"
]
}
],
"save_output" : true
"save_output" : false
}
]
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment