#97 Start test processing data in a SQL file

vincentnam · Sep 30, 2021 · dfb19e8 · dfb19e8
1 parent 7d8c6db
commit dfb19e8
Showing 1 changed file with 29 additions and 1 deletion.
diff --git a/process_zone/apache_airflow/dags/services/extract_transform_load_dump_sql.py b/process_zone/apache_airflow/dags/services/extract_transform_load_dump_sql.py
@@ -1,10 +1,38 @@
 import sys
 import MySQLdb
 import config
+from io import StringIO
 
 # Handling function about SQL dumps
 def extract_transform_load_sql_dump(swift_result, swift_container, swift_id, process_type):
 
+    s=str(swift_result,'utf-8').replace("\\n", "\n")
+    lines = StringIO()
+    lines.write(s)
+    lines.seek(0)
+    data = []
+    for line in lines:
+        line = line.replace("\n","")
+        data.append(line)
+
+    new_data = []
+    idInsert = 0
+    for d in data:
+        if("INSERT" in d):
+            idInsert += 1
+
+        if(";" in d and idInsert > 0):
+            idInsert += 1
+
+        if (idInsert != 0 and idInsert < 3):
+            new_data.append(d)
+            if(idInsert == 2):
+                idInsert = 0
+
+    processing_data = "\n".join(new_data)
+
+    print(processing_data)
+
     # Connect
     db = MySQLdb.connect(
         host=config.mariadb_host,
@@ -16,7 +44,7 @@ def extract_transform_load_sql_dump(swift_result, swift_container, swift_id, pro
     cursor = db.cursor()
 
     # Execute SQL select statement
-    cursor.execute(swift_result)
+    cursor.execute(processing_data)
 
     # Close the connection
     db.close()