adjust db executions

dataacquisition/ExportToDatabase.py

@@ -16,7 +16,7 @@ import configparser
 from api.GetAverageData import get_interpolation_data_for_point
 
 cfg = configparser.ConfigParser()
-cfg.read('./config.ini')
+cfg.read('../config.ini')
 assert "POSTGRES" in cfg, "missing POSTGRES in config.ini"
 param_postgres = cfg["POSTGRES"]
 param_interpol = cfg["INTERPOLATION"]
@@ -148,19 +148,20 @@ def create_matrix_data(cursor, amount_points):
 
 
 # Dumping all existing data from database. Inserting station data into database in bulk.
-def createInsertStatement(station_list):
+def createInsertStatement(cursor, station_list):
     # create INSERT INTO table (columns) VALUES('%s',...)
     # station_list.columns.astype(str)
 
     df_columns = list(station_list)
     columns = ['"' + column + '"' for column in df_columns]
-    
     columns = str(columns).replace('[', '').replace(']', '').replace("'", "").replace('\n', '').replace(' ', '')
-    values = "VALUES({})".format(",".join(["%s" for _ in station_list]))
-    #df_columns = str(df_columns).strip('[]')
+    value_template = "({})".format(",".join(["%s" for _ in df_columns]))
+    values = ','.join(cursor.mogrify(value_template, value.array).decode("utf-8") for i, value in station_list.iterrows())
     
+    df_columns = str(df_columns).strip('[]')
+    station_list = station_list.round(decimals=3)
     # create INSERT INTO table (columns) VALUES('%s',...)
-    insert_stmt = """INSERT INTO {} ({}) {}""".format('stations', columns, values)
+    insert_stmt = """INSERT INTO {} ({}) VALUES {};""".format('stations', columns, values)
     
     return insert_stmt
 
@@ -188,11 +189,12 @@ def insert_data(station_list, cursor):
 
         # create INSERT INTO table (columns) VALUES('%s',...)
         #insert_stmt = """INSERT INTO {} ({}) {}""".format('stations', columns, values)
-        insert_stmt = createInsertStatement(station_list)
+        insert_stmt = createInsertStatement(cursor, station_list)
         
-        psycopg2.extras.execute_batch(cursor, insert_stmt, station_list.values)
+        #psycopg2.extras.execute_batch(cursor, insert_stmt, station_list.values)
+        cursor.execute(insert_stmt)
         
-    print('Done')
+    print('Inserting data into database Done')
 
 
 def export(station_list):
@@ -209,21 +211,24 @@ def export(station_list):
         connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)  # Needs to be in AUTOCOMMIT mode for creating database
         with connection.cursor() as cursor:
             station_exists = check_for_stations_existence(cursor, "stations")
-        
-        connection = psycopg2.connect(database=db_name, user=user, password=pw, host=param_postgres["host"], port=port)
-        with connection.cursor() as cursor:
+            
             if station_exists == False: 
-                create_table(station_list, cursor)
+                create_table(station_list, cursor)        
             
-            #insert_data(station_list, cursor)
-            #amount_points = insert_empty_matrix_into_db(cursor)
+            insert_data(station_list, cursor)
+            amount_points = insert_empty_matrix_into_db(cursor)
     
-        # connection = psycopg2.connect(database=db_name, user=user, password=pw, host=param_postgres["host"], port=port,
-        #                      keepalives=1, keepalives_idle=30, keepalives_interval=10, keepalives_count=5)
-        # with connection.cursor() as cursor:
-        #     create_matrix_data(cursor, amount_points)
+            # connection = psycopg2.connect(database=db_name, user=user, password=pw, host=param_postgres["host"], port=port,
+            #                       keepalives=1, keepalives_idle=30, keepalives_interval=10, keepalives_count=5)
+            with connection.cursor() as cursor:
+                create_matrix_data(cursor, amount_points)
+        
+        print('Installation successful. You can run the api.py now.')
+    except(Exception, psycopg2.DatabaseError) as error:
+        print(error)
     
     finally:
         if connection:
             connection.close()
-    print('Installation successful. You can run the api.py now.')
+    
+    

dataacquisition/test_ExportToDatabase.py

@@ -13,12 +13,11 @@ import pickle
 import dataacquisition.ExportToDatabase as cut
 import configparser
 import psycopg2
-import time
 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
 from psycopg2 import sql
 
 cfg = configparser.ConfigParser()
-cfg.read('./config.ini')
+cfg.read('../config.ini')
 
 assert "POSTGRES" in cfg, "missing POSTGRES in config.ini"
 
@@ -122,7 +121,7 @@ class TestExportToDatabase(unittest.TestCase):
                 connection.close()
             
             # test
-            with open("./dataacquisition/pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
+            with open("./pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
                 stationList = pickle.load(pickleFile)
                 stationList = stationList.loc[stationList['country']=="Germany"]
                 stationList.columns = stationList.columns.astype(str)
@@ -153,23 +152,25 @@ class TestExportToDatabase(unittest.TestCase):
                 connection.close()
             self.assertRaises(error.message)
             print("Test FAILED__")
-    
+        
     def testCreateInsertStatement(self):
         print("__Test CreateInsertStatement")
-        with open("./dataacquisition/pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
+        with open("./pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
             stationList = pickle.load(pickleFile)
             stationList = stationList.loc[stationList['country']=="Germany"]
+            stationList = stationList[:2]
             stationList["station_id"] = stationList.index
             stationList.columns = stationList.columns.astype(str)
-            
-            insert_stmt = cut.createInsertStatement(stationList)
-            
-            self.assertTrue(insert_stmt.startswith("INSERT INTO"))
-            self.assertTrue(insert_stmt.endswith("%s)"))
+            connection = psycopg2.connect(dbname=param_postgres["dbName"], user=param_postgres["user"], password=param_postgres["password"], host=param_postgres["host"], port=param_postgres["port"])
+            with connection.cursor() as cursor:
+                insert_stmt = cut.createInsertStatement(cursor, stationList)
+                
+                self.assertTrue(insert_stmt.startswith("INSERT INTO"))
+                self.assertTrue(insert_stmt.endswith("');"))
     
     def _testIntegrationTest(self):
         print("__Test Integrationtest")
-        with open("./dataacquisition/pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
+        with open("./pickle/stationList_with_temperature.pickle", "rb") as pickleFile:
             stationList = pickle.load(pickleFile)
             stationList = stationList.loc[stationList['country']=="Germany"]
             # cut.drop_db(param_postgres['dbName']);