#87 - Handling SQL dump in Airflow

process_zone/apache_airflow/dags/config.py.sample

@@ -18,3 +18,9 @@ org_influxdb = ""
 bucket_influxdb = ""
 url_influxdb = "http://127.0.0.1:8086"
 influxdb_measurement = "value_units"
+
+# MariaDB
+mariadb_host=""
+mariadb_user=""
+mariadb_passwd=""
+mariadb_database=""

process_zone/apache_airflow/dags/data-processing-upload.py

@@ -9,6 +9,7 @@
 from services import extract_transform_load_time_series_csv
 from services import extract_transform_load_time_series_json
 from services import extract_transform_load_images
+from services import extract_transform_load_dump_sql
 from services import typefile
 import tempfile
 import base64
@@ -95,6 +96,7 @@ def get_swift_object(*args, **kwargs):
                 process_type = "time_series_csv"
                 # Json parsing
                 processed_data = extract_transform_load_time_series_csv(data_file, swift_container, swift_id, process_type)
+
 
     # Compare filetype
     if "image/" in content_type :
@@ -110,6 +112,11 @@ def get_swift_object(*args, **kwargs):
         # Json parsing
         processed_data = extract_transform_load_time_series_csv(swift_result, swift_container, swift_id, process_type)
 
+    if "application/sql" in content_type:
+        process_type = "sql_dump"
+        # Json parsing
+        processed_data = extract_transform_load_dump_sql(swift_result, swift_container, swift_id, process_type)
+
     # Handled data
     return processed_data
 

process_zone/apache_airflow/dags/services/extract_transform_load_dump_sql.py

@@ -0,0 +1,23 @@
+import MySQLdb
+import config
+
+# Handling function about SQL dumps
+def extract_transform_load_sql_dump(swift_result, swift_container, swift_id, process_type):
+
+    # Connect
+    db = MySQLdb.connect(
+        host=config.mariadb_host,
+        user=config.mariadb_user,
+        passwd=config.mariadb_passwd,
+        db=config.mariadb_database
+    )
+
+    cursor = db.cursor()
+
+    # Execute SQL select statement
+    cursor.execute(swift_result)
+
+    # Close the connection
+    db.close()
+
+sys.modules[__name__] = extract_transform_load_sql_dump