[06.02.20] Updated to work with graphql backend

configuration/dbgateway.env

@@ -0,0 +1,2 @@
+URI="http://localhost"
+PORT="8080"

src/pricing/collector.py

@@ -3,13 +3,15 @@
 import sys, json, os
 from datetime import datetime, timedelta
 
+from time import sleep
+
 from pricing.bitfinex import bitfinexPublicTicker, bitfinexHighLowVol
 from pricing.coinbase import coinbasePublicTicker
 from pricing.gemini import geminiPublicTicker, geminiHighLowVol, geminiOpenClose
 
-from util.databaseConnect import connect, closeConnection
+from utils.databaseConnect import send
 
-btc_usd="resources/sql/V1_INSERT_NEW_PRICE_RECORD_BTC.sql"
+btc_usd="resources/sql/V1_INSERT_NEW_PRICE_RECORD_BTC.graphql"
 
 def getInsertForType(type):
     if type == "btc_usd":
@@ -73,15 +75,16 @@ def getHighLowVol(type):
 
     return high, low, vol
 
-def saveToDatabase(type, timestamp, av_price, high, low, vol, open, close):
+def sendToGateway(type, timestamp, av_price, high, low, vol, open, close):
     try:
-        cur = connect()
+        with open(btc_usd, 'r') as queryFile:
+            data = queryFile.read()
+        query = data % (type, timestamp, av_price, high, low, vol, open, close)
 
-        sql = getInsertForType(type)
+        status, response = send(query)
 
-        cur.execute(sql, (type, timestamp, av_price, high, low, open, close, vol))
+        print("Status: ", status)
-
+        print("Response: ", response)
-        closeConnection(cur)
         return True
     except BaseException as exception:
         print("Error: %s" % str(exception))
@@ -103,4 +106,4 @@ def collector(type):
         high, low, vol = getHighLowVol(type)
         open, close = getOpenClose(type)
 
-        saveToDatabase(type, timestamp, av_price, high, low, vol, open, close)
+        sendToGateway(type, timestamp, av_price, high, low, vol, open, close)

src/resources/queries/V1_INSERT_NEW_PRICE_RECORD_BTC.graphql

@@ -0,0 +1 @@
+mutation { createBtc(createdDate: %s, type: %s, average_price: %f, high_price: %f, low_price: %f, open_price: %f, close_price: %f, volume: %f){ id } }

src/resources/sql/V1_INSERT_NEW_PRICE_RECORD_BTC.sql

@@ -1,2 +0,0 @@
-INSERT INTO btc_price(timestamp, symbol, av_price, h_price, l_price, o_price, c_price, volume)
-VALUES(%s, %s, %f, %f, %f, %f, %f, %f);

src/utils/databaseConfig.py

@@ -1,17 +0,0 @@
-#!/usr/bin/env python
-
-from configparser import ConfigParser
-
-def config(filename='configuration/database.ini', section='postgresql'):
-    parser = ConfigParser()
-    parser.read(filename)
-
-    db = {}
-    if parser.has_section(section):
-        params = parser.items(section)
-        for param in params:
-            db[param[0]] = param[1]
-    else:
-        raise Exception('Section {0} not found in the {1} file'.format(section, filename))
-
-    return db

src/utils/databaseConnect.py

@@ -1,39 +1,26 @@
 #!/usr/bin/env python
 
-import psycopg2
+import requests, os, json
-from utils.databaseConfig import config
 
-def connect():
+from dotenv import load_dotenv
-    """ Connect to the PostgreSQL database server """
+from pathlib import Path  # python3 only
-    conn = None
+env_path = Path('.') / 'configuration/dbgateway.env.env'
-    try:
+load_dotenv(dotenv_path=env_path)
-        # read connection parameters
-        params = config()
 
-        # connect to the PostgreSQL server
+class keys():
-        print('Connecting to the PostgreSQL database...')
-        conn = psycopg2.connect(**params)
 
-        # create a cursor
+    def __init__(self):
-        cur = conn.cursor()
+        self.uri = os.getenv('URI')
+        self.port = os.getenv("PORT")
 
-        # execute a statement
+def send(query):
-        print('PostgreSQL database version:')
-        cur.execute('SELECT version()')
 
-        # display the PostgreSQL database server version
+    uri = keys().uri + ":"+ keys().port + "/graphql"
-        db_version = cur.fetchone()
+    headers = {'Content-type': 'application/graphql'}
-        print(db_version)
 
-        # close the communication with the PostgreSQL
+    response = requests.post(uri, data=query, headers=headers)
-        # cur.close()
-        return cur;
-    except (Exception, psycopg2.DatabaseError) as error:
-        print(error)
-    finally:
-        if conn is not None:
-            conn.close()
-            print('Database connection closed.')
 
-def closeConnection(cur):
+    statusCode = json.loads(response.status_code)
-    cur.close()
+    response = json.loads(response.text)
+
+    return statusCode, response