59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
import mysql.connector
|
|
import pydapper
|
|
import logging
|
|
import json
|
|
import os
|
|
import sys
|
|
from BreCal.schemas import defs
|
|
config_path = None
|
|
|
|
def initPool(instancePath, connection_filename="connection_data_test.json"):
|
|
try:
|
|
global config_path
|
|
if(config_path == None):
|
|
config_path = os.path.join(instancePath,f'../../../secure/{connection_filename}') #connection_data_test.json');
|
|
|
|
config_path = "E:/temp/connection_data.json"
|
|
print (config_path)
|
|
|
|
if not os.path.exists(config_path):
|
|
print ('cannot find ' + os.path.abspath(config_path))
|
|
print("instance path", instancePath)
|
|
sys.exit(1)
|
|
|
|
f = open(config_path);
|
|
connection_data = json.load(f)
|
|
f.close()
|
|
|
|
conn_from_pool = mysql.connector.connect(**connection_data)
|
|
|
|
commands = pydapper.using(conn_from_pool)
|
|
data = commands.query("SELECT id from `user`")
|
|
print("DB connection successful")
|
|
conn_from_pool.close()
|
|
|
|
credentials_file = "email_credentials_test.json"
|
|
credentials_path = os.path.join(instancePath,f'../../../secure/{credentials_file}')
|
|
|
|
credentials_path = "E:/temp/email_credentials_devel.json"
|
|
|
|
if not os.path.exists(credentials_path):
|
|
print ('cannot find ' + os.path.abspath(credentials_path))
|
|
sys.exit(1)
|
|
|
|
f = open(credentials_path);
|
|
defs.email_credentials = json.load(f)
|
|
f.close()
|
|
|
|
except mysql.connector.PoolError as e:
|
|
logging.error(f"Failed to create connection pool: {e}")
|
|
print(e)
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
def getPoolConnection():
|
|
global config_path
|
|
f = open(config_path);
|
|
connection_data = json.load(f)
|
|
return mysql.connector.connect(**connection_data)
|