Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
B
bcom-tp-etl-transformation-pipelines
Project
Project
Details
Activity
Releases
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
general
bcom-tp-etl-transformation-pipelines
Commits
56c7f2a2
Commit
56c7f2a2
authored
Jul 25, 2023
by
Cristian Aguirre
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Update 25-07-23. Update DAG process. Version-2-inprogress
parent
55785c2d
Changes
13
Show whitespace changes
Inline
Side-by-side
Showing
13 changed files
with
423 additions
and
127 deletions
+423
-127
Database.py
dags/components/Databases/Database.py
+11
-0
Mysql.py
dags/components/Databases/Mysql.py
+10
-0
Oracle.py
dags/components/Databases/Oracle.py
+11
-0
Postgres.py
dags/components/Databases/Postgres.py
+11
-0
Extractor.py
dags/components/Extractor.py
+27
-49
Transformation.py
dags/components/Transformation.py
+1
-1
Utils.py
dags/components/Utils.py
+28
-36
dag_conf.yml
dags/dag_conf.yml
+6
-6
dag_transformacion_bcom.py
dags/dag_transformacion_bcom.py
+3
-3
dag_transformacion_tacomventas_promoresidencial.py
dags/dag_transformacion_tacomventas_promoresidencial.py
+1
-1
DataTypeEnum.py
dags/enums/DataTypeEnum.py
+2
-0
DataTypeOrmEnum.py
dags/enums/DataTypeOrmEnum.py
+3
-2
procedure_definition.json
dags/procedure_definition.json
+309
-29
No files found.
dags/components/Databases/Database.py
View file @
56c7f2a2
...
@@ -30,6 +30,7 @@ class Database:
...
@@ -30,6 +30,7 @@ class Database:
elif
db_type
==
DatabaseTypeEnum
.
ORACLE
.
value
:
elif
db_type
==
DatabaseTypeEnum
.
ORACLE
.
value
:
self
.
factory
=
Oracle
(
host
,
port
,
user
,
password
,
service
)
self
.
factory
=
Oracle
(
host
,
port
,
user
,
password
,
service
)
self
.
engine
=
None
self
.
engine
=
None
self
.
procedure_identifier
=
"execute"
def
get_basic_connection
(
self
):
def
get_basic_connection
(
self
):
connection
=
None
connection
=
None
...
@@ -73,3 +74,13 @@ class Database:
...
@@ -73,3 +74,13 @@ class Database:
logger
.
error
(
f
"Error creando tabla {model}. {e}"
)
logger
.
error
(
f
"Error creando tabla {model}. {e}"
)
finally
:
finally
:
return
save
return
save
def
generate_sql_procedure
(
self
,
command
:
str
)
->
str
:
response
=
""
try
:
command
=
command
.
lower
()
response
=
self
.
factory
.
generate_sql_procedure
(
command
,
self
.
procedure_identifier
)
except
Exception
as
e
:
logger
.
error
(
f
"Error generando comando sql para procedure. Comando: {command}. {e}"
)
finally
:
return
response
dags/components/Databases/Mysql.py
View file @
56c7f2a2
...
@@ -59,3 +59,13 @@ class Mysql:
...
@@ -59,3 +59,13 @@ class Mysql:
finally
:
finally
:
return
model
return
model
def
generate_sql_procedure
(
self
,
command
:
str
,
reserved_word
:
str
=
"execute"
)
->
str
:
response
=
""
try
:
command
=
command
.
replace
(
reserved_word
,
""
)
.
replace
(
";"
,
""
)
response
=
f
"call {command};"
logger
.
debug
(
"COMANDO MYSQL:"
,
response
)
except
Exception
as
e
:
logger
.
error
(
f
"Error generando comando sql para procedure Mysql. Comando: {command}. {e}"
)
finally
:
return
response
dags/components/Databases/Oracle.py
View file @
56c7f2a2
...
@@ -81,3 +81,14 @@ class Oracle:
...
@@ -81,3 +81,14 @@ class Oracle:
logger
.
error
(
f
"Error creando modelo dinámico en Oracle con nombre {tablename}. {type(e)}.{e}"
)
logger
.
error
(
f
"Error creando modelo dinámico en Oracle con nombre {tablename}. {type(e)}.{e}"
)
finally
:
finally
:
return
model
return
model
def
generate_sql_procedure
(
self
,
command
:
str
,
reserved_word
:
str
=
"execute"
)
->
str
:
response
=
""
try
:
command
=
command
.
replace
(
reserved_word
,
""
)
.
replace
(
";"
,
""
)
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}"
)
finally
:
return
response
dags/components/Databases/Postgres.py
View file @
56c7f2a2
...
@@ -59,3 +59,14 @@ class Postgres:
...
@@ -59,3 +59,14 @@ class Postgres:
logger
.
error
(
f
"Error creando modelo dinámico en Postgres con nombre {tablename}. {type(e)}. {e}"
)
logger
.
error
(
f
"Error creando modelo dinámico en Postgres con nombre {tablename}. {type(e)}. {e}"
)
finally
:
finally
:
return
model
return
model
def
generate_sql_procedure
(
self
,
command
:
str
,
reserved_word
:
str
=
"execute"
)
->
str
:
response
=
""
try
:
command
=
command
.
replace
(
reserved_word
,
""
)
.
replace
(
";"
,
""
)
response
=
f
"call {command};"
logger
.
debug
(
"COMANDO POSTGRES:"
,
response
)
except
Exception
as
e
:
logger
.
error
(
f
"Error generando comando sql para procedure Postgres. Comando: {command}. {e}"
)
finally
:
return
response
dags/components/Extractor.py
View file @
56c7f2a2
...
@@ -84,65 +84,43 @@ def on_success_extractor(context) -> None:
...
@@ -84,65 +84,43 @@ def on_success_extractor(context) -> None:
def
extract_from_source
(
command
:
str
,
source_conn
,
intern_conn
,
chunksize
:
int
,
**
kwargs
):
def
extract_from_source
(
command
:
str
,
source_conn
,
intern_conn
,
chunksize
:
int
,
**
kwargs
):
extract_type
=
command
[
0
]
task
=
kwargs
[
'ti'
]
extract_type
=
command
[
0
]
.
split
(
"|"
)[
0
]
command
=
command
[
1
]
command
=
command
[
1
]
source_engine
=
source_conn
.
engine
source_engine
=
source_conn
.
engine
command_for_create
=
command
definitions
=
task
.
xcom_pull
(
task_ids
=
"SCRIPTS-EXTRACTOR"
,
key
=
"EXTRACTION-DEFINITION-JSON"
)
is_tablename
=
False
logger
.
info
(
f
"Definiciones desde json: {definitions}"
)
multiple
=
select_multiple
(
command
)
tablename
=
multiple
[
"tablename"
]
is_procedure
=
False
if
extract_type
==
"PROCEDURE"
:
is_procedure
=
True
columns_name
=
[]
columns_name
=
[]
if
extract_type
.
startswith
(
OperationTypeEnum
.
PROCEDURE
.
value
):
# Create the model with the procedure descriptor
# Create the model with the procedure descriptor
if
command
.
replace
(
" "
,
""
)
.
lower
()
.
find
(
"|begin"
)
!=
-
1
:
tablename
=
command
[:
command
.
find
(
"|"
)]
else
:
proc_name
=
command
[
len
(
"begin"
):
command
.
rfind
(
"end"
)]
tablename
=
proc_name
.
strip
()
.
replace
(
";"
,
""
)
task
=
kwargs
[
'ti'
]
procedures
=
task
.
xcom_pull
(
task_ids
=
"SCRIPTS-EXTRACTOR"
,
key
=
"PROCEDURE-JSON"
)
model
=
None
model
=
None
for
procedure
in
procedure
s
:
for
procedure
in
definition
s
:
if
procedure
[
"procedure_
identifier"
]
!=
tablename
:
if
procedure
[
"
identifier"
]
!=
tablename
:
continue
continue
model
=
generateModel
(
tablename
,
procedure
[
"fields"
])
indexes
=
[]
columns_name
=
[
field
[
"identifier"
]
for
field
in
procedure
[
"fields"
]]
if
"indexes"
in
procedure
.
keys
():
is_tablename
=
True
indexes
=
procedure
[
"indexes"
]
model
=
generateModel
(
tablename
,
procedure
[
"fields"
],
indexes
)
columns_name
=
[
field
[
"name"
]
for
field
in
procedure
[
"fields"
]]
if
isinstance
(
model
,
type
(
None
)):
if
isinstance
(
model
,
type
(
None
)):
raise
AssertionError
(
"Descripción del procedure en el json descriptor no encontraddo"
)
raise
AssertionError
(
f
"Definición del extracción para {tablename} en el json-descriptor no encontraddo"
)
else
:
if
source_conn
.
db_type
==
DatabaseTypeEnum
.
ORACLE
.
value
and
\
not
extract_type
.
startswith
(
OperationTypeEnum
.
PROCEDURE
.
value
):
command_for_create
=
f
"SELECT * FROM ({command_for_create}) OFFSET 1 ROWS FETCH NEXT 1 ROWS ONLY"
else
:
command_words
=
command_for_create
.
split
(
" "
)
if
command_words
[
-
2
]
.
lower
()
!=
"limit"
:
command_for_create
+=
" limit 1"
columns
=
[]
with
source_engine
.
connect
()
as
connection
:
final_command
=
command_for_create
if
final_command
.
replace
(
" "
,
""
)
.
lower
()
.
find
(
"|select"
)
!=
-
1
:
final_command
=
final_command
[
final_command
.
find
(
"select"
):]
result
=
connection
.
execute
(
final_command
)
fields
=
result
.
cursor
.
description
for
field
in
fields
:
name
,
type_code
,
length
,
presicion
=
field
[
0
],
field
[
1
],
field
[
3
],
field
[
5
]
columns
.
append
((
name
,
type_code
,
length
,
presicion
))
logger
.
debug
(
f
"Columnas procesadas: {columns}"
)
multiple
=
select_multiple
(
command
)
tablename
=
multiple
[
"tablename"
]
model
=
source_conn
.
create_model
(
tablename
,
columns
)
try
:
try
:
create
=
intern_conn
.
create_table
(
model
)
create
=
intern_conn
.
create_table
(
model
)
if
create
:
if
create
:
logger
.
info
(
f
"Creado correctamente la tabla {tablename}. Creado?: {create}"
)
logger
.
info
(
f
"Creado correctamente la tabla {tablename}. Creado?: {create}"
)
else
:
else
:
raise
AssertionError
(
f
"Error creando tabla {tablename}"
)
raise
AssertionError
(
f
"Error creando tabla {tablename}"
)
if
is_procedure
:
if
is_tablename
:
command
=
command
[
len
(
tablename
+
"|"
):]
command
=
command
[
len
(
tablename
+
":"
):]
temp_connection
=
source_conn
.
get_basic_connection
()
temp_connection
=
source_conn
.
get_basic_connection
()
cursor
=
temp_connection
.
cursor
()
cursor
=
temp_connection
.
cursor
()
cursor
.
execute
(
command
)
final_command
=
source_conn
.
generate_sql_procedure
(
command
)
logger
.
debug
(
f
"FINAL COMMAND: {final_command}"
)
cursor
.
execute
(
final_command
)
for
resultSet
in
cursor
.
getimplicitresults
():
for
resultSet
in
cursor
.
getimplicitresults
():
data
=
[]
data
=
[]
for
row
in
resultSet
:
for
row
in
resultSet
:
...
@@ -201,7 +179,7 @@ def get_select_from_xcom(**kwargs):
...
@@ -201,7 +179,7 @@ def get_select_from_xcom(**kwargs):
for
select
in
xcom_selects
:
for
select
in
xcom_selects
:
tablename
=
select_multiple
(
select
)[
"tablename"
]
tablename
=
select_multiple
(
select
)[
"tablename"
]
if
tasks
[
"status"
]
==
ProcessStatusEnum
.
SUCCESS
.
value
or
tablename
not
in
success_tasks
:
if
tasks
[
"status"
]
==
ProcessStatusEnum
.
SUCCESS
.
value
or
tablename
not
in
success_tasks
:
final_selects
.
append
((
key
,
select
))
final_selects
.
append
((
key
,
select
,
))
logger
.
info
(
f
"Comandos para la extracción: {final_selects}"
)
logger
.
info
(
f
"Comandos para la extracción: {final_selects}"
)
Variable
.
set
(
key
=
'SELECTS'
,
value
=
final_selects
,
serialize_json
=
True
)
Variable
.
set
(
key
=
'SELECTS'
,
value
=
final_selects
,
serialize_json
=
True
)
return
[[
item
]
for
item
in
final_selects
]
return
[[
item
]
for
item
in
final_selects
]
...
...
dags/components/Transformation.py
View file @
56c7f2a2
...
@@ -82,7 +82,7 @@ def transformations(xcom_commands: str, intern_conn):
...
@@ -82,7 +82,7 @@ def transformations(xcom_commands: str, intern_conn):
logger
.
info
(
f
"Ejecutando transformaciones del script {script_name}"
)
logger
.
info
(
f
"Ejecutando transformaciones del script {script_name}"
)
with
engine
.
connect
()
as
connection
:
with
engine
.
connect
()
as
connection
:
for
command
in
commands
:
for
command
in
commands
:
logger
.
debug
(
f
"Ejecutando comando de transformación: {command}"
)
logger
.
info
(
f
"Ejecutando comando de transformación: {command}"
)
_
=
connection
.
execute
(
command
)
_
=
connection
.
execute
(
command
)
...
...
dags/components/Utils.py
View file @
56c7f2a2
...
@@ -93,6 +93,7 @@ def update_dict_with_catalogs(data_dict: Dict[str, Any], data: Dict[str, Any], c
...
@@ -93,6 +93,7 @@ def update_dict_with_catalogs(data_dict: Dict[str, Any], data: Dict[str, Any], c
def
update_sql_commands
(
dataset
:
List
[
Tuple
[
str
,
str
]],
label_tablename
:
str
)
->
List
[
Tuple
[
str
,
List
[
str
]]]:
def
update_sql_commands
(
dataset
:
List
[
Tuple
[
str
,
str
]],
label_tablename
:
str
)
->
List
[
Tuple
[
str
,
List
[
str
]]]:
result
=
[]
result
=
[]
allowed_commands
=
[
"create"
,
"update"
,
"delete"
,
"select"
,
"alter"
,
"drop"
,
"begin"
,
"commit"
]
comments
=
[
CommentsScriptEnum
[
item
]
.
value
for
item
in
CommentsScriptEnum
.
_member_names_
]
comments
=
[
CommentsScriptEnum
[
item
]
.
value
for
item
in
CommentsScriptEnum
.
_member_names_
]
try
:
try
:
for
row
in
dataset
:
for
row
in
dataset
:
...
@@ -109,7 +110,7 @@ def update_sql_commands(dataset: List[Tuple[str, str]], label_tablename: str) ->
...
@@ -109,7 +110,7 @@ def update_sql_commands(dataset: List[Tuple[str, str]], label_tablename: str) ->
final_item
=
item
final_item
=
item
if
item
.
lower
()
.
strip
()
.
find
(
label_tablename
.
lower
()
.
strip
()
+
":"
)
!=
-
1
:
if
item
.
lower
()
.
strip
()
.
find
(
label_tablename
.
lower
()
.
strip
()
+
":"
)
!=
-
1
:
init_index
=
item
.
lower
()
.
strip
()
.
index
(
label_tablename
.
lower
()
.
strip
()
+
":"
)
init_index
=
item
.
lower
()
.
strip
()
.
index
(
label_tablename
.
lower
()
.
strip
()
+
":"
)
table_name
=
item
.
replace
(
" "
,
""
)
.
strip
()[
init_index
+
5
:]
.
strip
()
table_name
=
item
.
replace
(
" "
,
""
)
.
strip
()[
init_index
+
len
(
label_tablename
)
:]
.
strip
()
add_next
=
True
add_next
=
True
elif
item
!=
""
:
elif
item
!=
""
:
if
add_next
:
if
add_next
:
...
@@ -117,38 +118,21 @@ def update_sql_commands(dataset: List[Tuple[str, str]], label_tablename: str) ->
...
@@ -117,38 +118,21 @@ def update_sql_commands(dataset: List[Tuple[str, str]], label_tablename: str) ->
add_next
=
False
add_next
=
False
final_item
=
item
.
strip
()
final_item
=
item
.
strip
()
table_name
=
""
table_name
=
""
if
final_item
.
strip
()[:
2
]
in
comments
and
(
"update "
in
final_item
.
lower
()
or
"delete "
in
final_item
.
lower
()
or
if
any
([
command
in
final_item
.
lower
()
for
command
in
allowed_commands
]):
"alter"
in
final_item
.
lower
()
or
"create"
in
final_item
.
lower
()
or
if
final_item
.
strip
()[:
2
]
in
comments
:
"drop"
in
final_item
.
lower
()):
init_indexes
=
[
final_item
.
lower
()
.
find
(
command
)
for
command
in
allowed_commands
trans_index
=
final_item
.
lower
()
.
find
(
"update"
)
if
final_item
.
lower
()
.
find
(
command
)
!=
-
1
]
if
trans_index
!=
-
1
:
if
len
(
init_indexes
)
>
0
:
final_item
=
final_item
[
trans_index
:]
init_index
=
init_indexes
[
0
]
delete_index
=
final_item
.
lower
()
.
find
(
"delete"
)
final_item
=
final_item
[
init_index
:]
if
delete_index
!=
-
1
:
final_item
=
final_item
[
delete_index
:]
alter_index
=
final_item
.
lower
()
.
find
(
"alter"
)
if
alter_index
!=
-
1
:
final_item
=
final_item
[
alter_index
:]
create_index
=
final_item
.
lower
()
.
find
(
"create"
)
if
create_index
!=
-
1
:
final_item
=
final_item
[
create_index
:]
drop_index
=
final_item
.
lower
()
.
find
(
"drop"
)
if
drop_index
!=
-
1
:
final_item
=
final_item
[
drop_index
:]
final_item
=
final_item
.
replace
(
"
%
"
,
"
%%
"
)
final_item
=
final_item
.
replace
(
"
%
"
,
"
%%
"
)
final_data
.
append
(
final_item
)
final_data
.
append
(
final_item
)
final_data
=
[
item
.
replace
(
"
\t
"
,
""
)
for
item
in
final_data
if
item
!=
""
and
(
"select"
in
item
.
lower
()
or
final_data
=
[
item
.
replace
(
"
\t
"
,
""
)
for
item
in
final_data
if
item
!=
""
and
"update"
in
item
.
lower
()
or
all
([
not
item
.
strip
()
.
startswith
(
comment
)
for
comment
in
comments
])]
"delete"
in
item
.
lower
()
or
"begin"
in
item
.
lower
()
or
"alter"
in
item
.
lower
()
or
"create"
in
item
.
lower
()
or
"drop"
in
item
.
lower
()
or
"commit"
in
item
.
lower
())]
result
.
append
((
row
[
0
],
final_data
))
result
.
append
((
row
[
0
],
final_data
))
logger
.
info
(
f
"Lista de comandos: {result}"
)
logger
.
info
(
f
"Lista de comandos: {result}"
)
except
Exception
as
e
:
except
Exception
as
e
:
raise
AssertionE
rror
(
f
"Error extrayendo comandos sql. {e}"
)
logger
.
e
rror
(
f
"Error extrayendo comandos sql. {e}"
)
finally
:
finally
:
return
result
return
result
...
@@ -157,9 +141,8 @@ def select_multiple(command: str) -> Dict[str, Any]:
...
@@ -157,9 +141,8 @@ def select_multiple(command: str) -> Dict[str, Any]:
response
=
{
'is_multiple'
:
False
,
'tablename'
:
''
}
response
=
{
'is_multiple'
:
False
,
'tablename'
:
''
}
tablename
=
""
tablename
=
""
no_procedure_init
=
"|select"
no_procedure_init
=
"|select"
procedure_init
=
[
"|begin"
]
try
:
try
:
if
command
.
lower
()
.
replace
(
" "
,
""
)
.
find
(
procedure_init
[
0
])
!
=
-
1
:
if
command
.
lower
()
.
replace
(
" "
,
""
)
.
find
(
no_procedure_init
)
=
=
-
1
:
response
[
"is_multiple"
]
=
True
response
[
"is_multiple"
]
=
True
tablename
=
command
[:
command
.
index
(
"|"
)]
.
strip
()
tablename
=
command
[:
command
.
index
(
"|"
)]
.
strip
()
response
[
"tablename"
]
=
tablename
response
[
"tablename"
]
=
tablename
...
@@ -217,21 +200,30 @@ def delete_temp_dir(module_name: str) -> bool:
...
@@ -217,21 +200,30 @@ def delete_temp_dir(module_name: str) -> bool:
return
drop
return
drop
def
generateModel
(
tablename
:
str
,
attributes
:
List
[
Dict
[
str
,
Any
]],
modelName
:
str
=
"TableModel"
):
def
generateModel
(
tablename
:
str
,
attributes
:
List
[
Dict
[
str
,
Any
]],
indexes
:
List
[
str
],
modelName
:
str
=
"TableModel"
):
default_precision
=
8
model
=
type
(
modelName
,
(
InsumoModel
,),
{
model
=
type
(
modelName
,
(
InsumoModel
,),
{
'__tablename__'
:
tablename
,
'__tablename__'
:
tablename
,
'__table_args__'
:
{
'extend_existing'
:
True
}
'__table_args__'
:
{
'extend_existing'
:
True
}
})
})
try
:
try
:
for
attribute
in
attributes
:
for
attribute
in
attributes
:
index
=
False
if
attribute
[
"name"
]
in
indexes
:
index
=
True
logger
.
debug
(
f
"attribute: {attribute}"
)
logger
.
debug
(
f
"attribute: {attribute}"
)
if
attribute
[
"datatype"
]
==
DataTypeEnum
.
TEXT
.
name
and
"maxLength"
in
attribute
.
keys
():
if
attribute
[
"datatype"
]
==
DataTypeEnum
.
TEXT
.
name
and
"maxLength"
in
attribute
.
keys
():
setattr
(
model
,
attribute
[
"identifier"
],
setattr
(
model
,
attribute
[
"name"
],
Column
(
DataTypeOrmEnum
[
attribute
[
"datatype"
]]
.
value
(
attribute
[
"maxLength"
])))
Column
(
DataTypeOrmEnum
[
attribute
[
"datatype"
]]
.
value
(
attribute
[
"maxLength"
]),
index
=
index
))
elif
attribute
[
"datatype"
]
==
DataTypeEnum
.
DECIMAL
.
name
:
precision
=
default_precision
if
"decimal_precision"
in
attribute
.
keys
():
precision
=
attribute
[
"decimal_precision"
]
setattr
(
model
,
attribute
[
"name"
],
Column
(
DataTypeOrmEnum
[
attribute
[
"datatype"
]]
.
value
(
38
,
precision
),
index
=
index
))
else
:
else
:
setattr
(
model
,
attribute
[
"
identifier
"
],
setattr
(
model
,
attribute
[
"
name
"
],
Column
(
DataTypeOrmEnum
[
attribute
[
"datatype"
]]
.
value
))
Column
(
DataTypeOrmEnum
[
attribute
[
"datatype"
]]
.
value
,
index
=
index
))
model
=
model
.
__table__
model
=
model
.
__table__
except
InvalidRequestError
as
e
:
except
InvalidRequestError
as
e
:
logger
.
debug
(
f
"InvalidRequestError. {e}"
)
logger
.
debug
(
f
"InvalidRequestError. {e}"
)
...
...
dags/dag_conf.yml
View file @
56c7f2a2
...
@@ -3,11 +3,11 @@ app:
...
@@ -3,11 +3,11 @@ app:
database
:
database
:
sources
:
sources
:
source1
:
source1
:
type
:
mysql
type
:
oracle
host
:
192.168.2
1.5
2
host
:
192.168.2
7.2
2
port
:
13306
port
:
21521
username
:
root
username
:
PRUEBABCOM2
password
:
root
password
:
admin
database
:
bd_tp_qa
database
:
bd_tp_qa
service
:
ORCLPDB1
service
:
ORCLPDB1
schema
:
public
schema
:
public
...
@@ -21,7 +21,7 @@ app:
...
@@ -21,7 +21,7 @@ app:
service
:
service
:
schema
:
schema
:
chunksize
:
8000
chunksize
:
8000
label_multiple_select
:
TABLE
label_multiple_select
:
TABLE
NAME
source_mask
:
select
# Sufijo (S)
source_mask
:
select
# Sufijo (S)
procedure_mask
:
procedure
# S
procedure_mask
:
procedure
# S
transformation_mask
:
transform
# S
transformation_mask
:
transform
# S
...
...
dags/dag_transformacion_bcom.py
View file @
56c7f2a2
...
@@ -7,7 +7,7 @@ from airflow import DAG
...
@@ -7,7 +7,7 @@ from airflow import DAG
from
airflow.operators.python
import
PythonOperator
from
airflow.operators.python
import
PythonOperator
from
airflow.utils.task_group
import
TaskGroup
from
airflow.utils.task_group
import
TaskGroup
from
components.Utils
import
update_sql_commands
from
components.Utils
import
update_sql_commands
from
components.Xcom
import
save_commands_to_xcom
,
delete_task_instances
from
components.Xcom
import
save_commands_to_xcom
from
components.S3Route
import
get_files_from_prefix
,
get_file_from_key
from
components.S3Route
import
get_files_from_prefix
,
get_file_from_key
from
components.Sensor
import
create_s3_sensor
from
components.Sensor
import
create_s3_sensor
from
components.Extractor
import
get_extract_task_group
from
components.Extractor
import
get_extract_task_group
...
@@ -25,7 +25,7 @@ logger = logging.getLogger()
...
@@ -25,7 +25,7 @@ logger = logging.getLogger()
DAG_NAME
=
"BCOM_DAG_TRANSFORMACIONES3"
DAG_NAME
=
"BCOM_DAG_TRANSFORMACIONES3"
# Change this path if is deployed in prod or dev
# Change this path if is deployed in prod or dev
MAIN_PATH
=
"/
roo
t/airflow/dags/"
MAIN_PATH
=
"/
op
t/airflow/dags/"
JSON_PROCEDURE_PATH
=
MAIN_PATH
+
"procedure_definition.json"
JSON_PROCEDURE_PATH
=
MAIN_PATH
+
"procedure_definition.json"
DEFAULT_ARGS
=
{
DEFAULT_ARGS
=
{
...
@@ -87,7 +87,7 @@ def save_procedure_json(json_path: str, task) -> None:
...
@@ -87,7 +87,7 @@ def save_procedure_json(json_path: str, task) -> None:
data
=
json
.
load
(
f
)
data
=
json
.
load
(
f
)
logger
.
info
(
f
"JSON-PROCEDURE {data}"
)
logger
.
info
(
f
"JSON-PROCEDURE {data}"
)
if
data
:
if
data
:
task
.
xcom_push
(
key
=
"
PROCEDURE
-JSON"
,
value
=
data
)
task
.
xcom_push
(
key
=
"
EXTRACTION-DEFINITION
-JSON"
,
value
=
data
)
except
Exception
as
e
:
except
Exception
as
e
:
logger
.
error
(
f
"Error leyendo y guardando archivo descriptor de procedure. {e}"
)
logger
.
error
(
f
"Error leyendo y guardando archivo descriptor de procedure. {e}"
)
...
...
dags/dag_transformacion_tacomventas_promoresidencial.py
View file @
56c7f2a2
...
@@ -234,7 +234,7 @@ def set_dag_1():
...
@@ -234,7 +234,7 @@ def set_dag_1():
from
yaml.loader
import
SafeLoader
from
yaml.loader
import
SafeLoader
# Cambiar conf_path dependiendo del ambiente, en prod usando k8 y contenedores usar /opt/airflow/dags/app_conf.yml
# Cambiar conf_path dependiendo del ambiente, en prod usando k8 y contenedores usar /opt/airflow/dags/app_conf.yml
# En desarrollo, cualquiera que apunte a su carpeta dags
# En desarrollo, cualquiera que apunte a su carpeta dags
conf_path
=
"/
roo
t/airflow/dags/app_conf.yml"
conf_path
=
"/
op
t/airflow/dags/app_conf.yml"
with
open
(
conf_path
)
as
f
:
with
open
(
conf_path
)
as
f
:
data
=
yaml
.
load
(
f
,
Loader
=
SafeLoader
)
data
=
yaml
.
load
(
f
,
Loader
=
SafeLoader
)
general_cnf
=
data
[
"general"
]
general_cnf
=
data
[
"general"
]
...
...
dags/enums/DataTypeEnum.py
View file @
56c7f2a2
...
@@ -8,5 +8,7 @@ class DataTypeEnum(enum.Enum):
...
@@ -8,5 +8,7 @@ class DataTypeEnum(enum.Enum):
NUMBER
=
int
NUMBER
=
int
TEXT
=
str
TEXT
=
str
DECIMAL
=
float
DECIMAL
=
float
BOOLEAN
=
bool
DATETIME
=
Timestamp
DATETIME
=
Timestamp
DATE
=
pd
.
Timestamp
DATE
=
pd
.
Timestamp
LONGTEXT
=
str
dags/enums/DataTypeOrmEnum.py
View file @
56c7f2a2
import
enum
import
enum
from
sqlalchemy
import
Integer
,
String
,
Date
,
DateTime
,
DECIMAL
from
sqlalchemy
import
Integer
,
String
,
Date
,
DateTime
,
DECIMAL
,
BOOLEAN
,
TEXT
class
DataTypeOrmEnum
(
enum
.
Enum
):
class
DataTypeOrmEnum
(
enum
.
Enum
):
...
@@ -8,4 +8,5 @@ class DataTypeOrmEnum(enum.Enum):
...
@@ -8,4 +8,5 @@ class DataTypeOrmEnum(enum.Enum):
DECIMAL
=
DECIMAL
DECIMAL
=
DECIMAL
DATE
=
Date
DATE
=
Date
DATETIME
=
DateTime
DATETIME
=
DateTime
BOOLEAN
=
BOOLEAN
LONGTEXT
=
TEXT
dags/procedure_definition.json
View file @
56c7f2a2
[{
[
"procedure_identifier"
:
"PROCEDURE_1"
,
{
"identifier"
:
"TACOMVENTAS"
,
"fields"
:
[
"fields"
:
[
{
{
"identifier
"
:
"CD_EMPRESA"
,
"name
"
:
"CD_EMPRESA"
,
"datatype"
:
"NUMBER"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
,
"decimal_precision"
:
0
"maxLength"
:
null
},
},
{
{
"identifier
"
:
"CD_FOLIO"
,
"name
"
:
"CD_FOLIO"
,
"datatype"
:
"TEXT"
,
"datatype"
:
"TEXT"
,
"decimal_precision"
:
null
,
"maxLength"
:
100
"maxLength"
:
100
},
},
{
{
"identifier"
:
"CD_CLIENTE"
,
"name"
:
"CD_CUENTA"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"NU_VENDEDOR"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"NU_ADDON"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"NB_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
200
},
{
"name"
:
"CD_CLIENTE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"NB_CLIENTE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
200
},
{
"name"
:
"FH_CONTRATACION"
,
"datatype"
:
"DATE"
},
{
"name"
:
"FH_ACTIVACION"
,
"datatype"
:
"DATE"
},
{
"name"
:
"FH_OPERACION"
,
"datatype"
:
"DATE"
},
{
"name"
:
"TP_SERVICIO"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"ST_CLIENTE"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"TP_PAGO"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
10
},
{
"name"
:
"NB_USUACARGA"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"FH_CARGA"
,
"datatype"
:
"DATE"
},
{
"name"
:
"NU_ANIO"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"NU_MES"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"NU_SEMANA"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"NU_COMISION"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"TP_PAGOANT"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"REGLA_APLICADA"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"AUX"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"PROMOCION"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
80
},
{
"name"
:
"EMPLEADOEMBAJADOR__C"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"CANAL_DE_VENTA__C"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"CANALREFERIDO__C"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
}
],
"indexes"
:
[
"CD_PAQUETE"
,
"NU_ADDON"
,
"CD_CLIENTE"
]
},
{
"identifier"
:
"PROMOCIONES_RESIDENCIAL"
,
"fields"
:
[
{
"name"
:
"EMPRESA"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"CUENTA"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"PLAN"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"NOMBRE_PRODUCTO"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"DESCR"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"TIPO_ADICION"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
250
},
{
"name"
:
"RENTACONIMPUESTOS"
,
"datatype"
:
"NUMBER"
},
{
"name"
:
"RENTASINIMPUESTOS"
,
"datatype"
:
"NUMBER"
},
{
"name"
:
"QUANTITY"
,
"datatype"
:
"NUMBER"
,
"decimal_precision"
:
0
},
{
"name"
:
"CREACION_PRODUCTO"
,
"datatype"
:
"DATE"
},
{
"name"
:
"INICIO_COBRO"
,
"datatype"
:
"DATE"
},
{
"name"
:
"FIN_COBRO"
,
"datatype"
:
"DATE"
},
{
"name"
:
"FIN_COMPRA"
,
"datatype"
:
"DATE"
},
{
"name"
:
"SERV_STATUS"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
20
},
{
"name"
:
"POID_TYPE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
200
},
{
"name"
:
"POID_PRODUCT"
,
"datatype"
:
"TEXT"
,
"datatype"
:
"TEXT"
,
"decimal_precision"
:
null
,
"maxLength"
:
50
"maxLength"
:
50
},
},
{
{
"identifier"
:
"FH_CONTRATACION"
,
"name"
:
"STATUS_PRODUCTO"
,
"datatype"
:
"DATE"
,
"datatype"
:
"TEXT"
,
"pattern"
:
"%d-%m-%y"
,
"decimal_precision"
:
null
,
"decimal_precision"
:
null
,
"maxLength"
:
null
"maxLength"
:
200
},
{
"name"
:
"SERVICIO"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
200
},
{
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
}
],
"indexes"
:
[
"CD_PAQUETE"
]
},
{
"identifier"
:
"CATALOGO_PROMOCIONES"
,
"fields"
:
[
{
"name"
:
"NOMBRE_PRODUCTO"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
}
}
]
]
},
},
{
{
"
procedure_identifier"
:
"procedure_prueba2
"
,
"
identifier"
:
"RELACION_PROMOCION_3PA2P
"
,
"fields"
:
[
"fields"
:
[
{
{
"identifier"
:
"col1"
,
"name"
:
"TRESP"
,
"datatype"
:
"DATE"
,
"datatype"
:
"TEXT"
,
"pattern"
:
"%Y-%m-%d"
,
"maxLength"
:
50
"decimal_precision"
:
null
,
"maxLength"
:
null
},
},
{
{
"identifier"
:
"col2"
,
"name"
:
"DOSP"
,
"datatype"
:
"TIME"
,
"datatype"
:
"TEXT"
,
"pattern"
:
"%H:%M:%S"
,
"maxLength"
:
50
"decimal_precision"
:
null
,
}
"maxLength"
:
null
]
},
{
"identifier"
:
"RELACION_POIDPAQUETE"
,
"fields"
:
[
{
"name"
:
"POID_PRODUCT"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
},
{
{
"identifier"
:
"col3"
,
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"DATETIME"
,
"datatype"
:
"TEXT"
,
"pattern"
:
"%Y-%m-%d %H:%M:%S"
,
"maxLength"
:
50
"decimal_precision"
:
null
,
}
"maxLength"
:
null
]
},
{
"identifier"
:
"RELACION_PAQINI_PAQFIN"
,
"fields"
:
[
{
"name"
:
"COD_PAQ_INI"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"COD_PAQ_FIN"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
}
]
},
{
"identifier"
:
"PAQUETES_NOPROMOCION"
,
"fields"
:
[
{
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
}
]
},
{
"identifier"
:
"PROCEDURE_1"
,
"fields"
:
[
{
"name"
:
"CD_FOLIO"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"CD_CUENTA"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
100
},
{
"name"
:
"CD_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
50
},
{
"name"
:
"NB_PAQUETE"
,
"datatype"
:
"TEXT"
,
"maxLength"
:
200
}
}
]
]
}
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment