Exchange Extract MYSQL Code

meltano-el/.gitignore

@@ -1,3 +0,0 @@
-/venv
-/.meltano
-.env

meltano-el/meltano.yml

@@ -1,7 +0,0 @@
-version: 1
-default_environment: dev
-environments:
-- name: dev
-- name: staging
-- name: prod
-project_id: 24ae7352-4d74-4ba6-988a-389fdd47eb74

meltano-el/output/.gitignore

@@ -1,2 +0,0 @@
-*
-!.gitignore

python_el/extract/exchange_extract/EXCHANGE_CONFIG.json

@@ -0,0 +1,6 @@
+{
+    "host":"127.0.0.1",
+    "user":"admin",
+    "password":"admin",
+    "database":"classicmodels"
+}

python_el/extract/exchange_extract/EXCHANGE_EXTRACT_SQL.json

@@ -0,0 +1,10 @@
+{
+ "customers"    :   "SELECT * FROM customers;",
+ "employees"    :   "SELECT * FROM employees;",
+ "offices"      :   "SELECT * FROM offices;",
+ "orderdetails" :   "SELECT * FROM orderdetails;",
+ "orders"       :   "SELECT * FROM orders;",
+ "payments"     :   "SELECT * FROM payments;",
+ "productlines" :   "SELECT * FROM productlines;",
+ "products"     :   "SELECT * FROM products;"
+}

python_el/extract/exchange_extract/exchange_extract_main.py

@@ -0,0 +1,90 @@
+from ntpath import join
+from operator import concat
+import pandas as pd
+import mysql.connector
+import json
+import os
+import csv
+
+def change_directory_to(path):
+  try:
+    current_directory = os.getcwd()
+    target_dir = concat(current_directory,path)
+    return os.chdir(target_dir)
+
+  except Exception as e:
+    print(e)
+    print("[FUNCTION_ERROR]-change_directory_to")
+
+
+
+def read_json_file(name):
+  try:
+    with open(name) as json_file:
+      output = json.load(json_file)
+    return output
+  except Exception as e:
+    print(e)
+    print("[FUNCTION_ERROR]-read_json_file:" , name)
+
+
+
+def connect_mysql(config):
+  mydb = mysql.connector.connect(
+  host      = config["host"],
+  user      = config["user"],
+  password  = config["password"],
+  database  = config["database"])
+  return mydb
+  
+
+
+def write_to_csv(path,dbname,tablename,cursor,data):
+  try:
+    filename = '/{}_{}.csv'.format(dbname, tablename)
+
+    headers = [col[0] for col in cursor.description] # get headers
+    data.insert(0, tuple(headers))
+    fp = open(concat(path,filename), 'w', newline = '') 
+    myFile = csv.writer(fp)
+    myFile.writerows(data)
+    fp.close()
+
+  except Exception as e:
+    print(e)
+    print("[FUNCTION_ERROR]-write_to_csv:" , filename)
+
+
+#Get all the Compulsory File
+CURRENT_DIRECTORY_PATH = "\\python_el\\extract\\exchange_extract"
+MYSQL_CONFIG_FILE = "EXCHANGE_CONFIG.json"
+SQL_FILE = "EXCHANGE_EXTRACT_SQL.json"
+CSV_STORAGE_PATH = "exchange_temp_extract_csv/"
+
+#Change directory & open all file
+change_directory_to(CURRENT_DIRECTORY_PATH)
+MYSQL_CONFIG = read_json_file(MYSQL_CONFIG_FILE)
+SQL_DICT = read_json_file(SQL_FILE)
+
+
+#Connect DB
+mydb = connect_mysql(MYSQL_CONFIG)
+mycursor = mydb.cursor()
+
+
+for tablename,sql in SQL_DICT.items():
+
+  try:
+    mycursor.execute(sql)
+    myresult = mycursor.fetchall()
+    dbname = mydb.database
+    write_to_csv(CSV_STORAGE_PATH,dbname,tablename,mycursor,myresult)
+  except Exception as e: 
+    print(e)
+    print("[ERROR]-SQL ERROR",tablename,sql)
+
+  
+mydb.close()
+
+
+