freezed/ocp5
freezed/ocp5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Packs database script in a class

Browse Source 
Adds little tests
This commit is contained in:
Fred Z 2018-07-30 07:25:21 +02:00
parent cff82380b8
commit ca821728ae
1 changed files with 71 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
db.py
View File

@ -12,74 +12,93 @@ Connect to DB
import pymysql.cursors import pymysql.cursors
from config import DB_CONFIG from config import DB_CONFIG
DB_NOT_FOUND = True
class Db():
# FUNCTION
def sql_create_db(filename=DB_CONFIG['file']):
""" """
Get the SQL instruction to create the DB for file Class doc
:return: a list of each SQL query whithout the trailing ";" :Test:
>>> test = Db()
:Tests: >>> print(test.message)
>>> sql_create_db('wronq_file.sql') DB «localhost» exist, using it
File load error : wronq_file.sql
False
""" """
from os import path
# Loading file def __init__(self):
if path.isfile(filename) is False: """
print("File load error : {}".format(filename)) Class initialiser
return False """
self.DB_NOT_FOUND = True
self.MSG = {
"request": "-- REQUEST #{} : --{}",
"database": "DB «{}» exist, using it"}
self.message = ""
else: # Connect to the database
with open(filename, "r") as sql_file: self.connection = pymysql.connect(
# Split file in list host=DB_CONFIG['host'],
ret = sql_file.read().split(';') user=DB_CONFIG['user'],
# drop last empty entry password=DB_CONFIG['pass'],
ret.pop() charset=DB_CONFIG['char'],
return ret cursorclass=pymysql.cursors.DictCursor)
try:
with self.connection.cursor() as cursor:
# Show DB
cursor.execute("SHOW DATABASES")
db_list = cursor.fetchall()
# WORK for idx, unused in enumerate(db_list):
# Connect to the database
CONNECTION = pymysql.connect(host=DB_CONFIG['host'],
user=DB_CONFIG['user'],
password=DB_CONFIG['pass'],
charset=DB_CONFIG['char'],
cursorclass=pymysql.cursors.DictCursor)
try: if DB_CONFIG['db'] in db_list[idx].values():
with CONNECTION.cursor() as cursor: DB_NOT_FOUND = False
# Show DB cursor.execute("USE {}".format(DB_CONFIG['db']))
cursor.execute("SHOW DATABASES") self.message = self.MSG['database'].format(
db_list = cursor.fetchall() DB_CONFIG['host'])
for idx, val in enumerate(db_list): # No DB, create it
if DB_NOT_FOUND:
request_list = self.get_sql_from_file()
if DB_CONFIG['db'] in db_list[idx].values(): if request_list is not False:
DB_NOT_FOUND = False
cursor.execute("USE {}".format(DB_CONFIG['db']))
print('DB exist : ready to use it.')
# No DB, create it for idx, sql_request in enumerate(request_list):
if DB_NOT_FOUND: self.message = self.MSG['request'].format(
request_list = sql_create_db() idx, sql_request)
cursor.execute(sql_request + ';')
if request_list is not False: except pymysql.err.OperationalError as except_detail:
print("DB error: «{}»".format(except_detail))
for idx, sql_request in enumerate(request_list): @staticmethod
# print(sql_request + ';') def get_sql_from_file(filename=DB_CONFIG['file']):
print("-- REQUEST #{} : --{}".format(idx, sql_request)) """
cursor.execute(sql_request + ';') Get the SQL instruction from a file
except pymysql.err.OperationalError as except_detail: :return: a list of each SQL query whithout the trailing ";"
print("DB error: «{}»".format(except_detail))
finally: :Tests:
CONNECTION.close() >>> Db.get_sql_from_file('wronq_file.sql')
File load error : wronq_file.sql
False
"""
from os import path
# File did not exists
if path.isfile(filename) is False:
print("File load error : {}".format(filename))
return False
else:
with open(filename, "r") as sql_file:
# Split file in list
ret = sql_file.read().split(';')
# drop last empty entry
ret.pop()
return ret
def __del__(self):
""" Object destruction"""
self.connection.close()
if __name__ == "__main__": if __name__ == "__main__":