File size: 4,704 Bytes
0b4e916 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
import time
import json
from typing import Any, Callable, Union
import mysql.connector
import subprocess
class CaesarSQL:
def __init__(self,host:str='localhost',user:str='root',password:str="temp123") -> None:
# Makes SQL connection to remote server.
self.connection = mysql.connector.connect(
host=host,
user=user,
password = password,
autocommit=True
)
def check_exists(self,result :Any):
# Checks if an entity exists from an SQL Command.
try:
if len(result) == 0:
return False
else:
return True
except Exception as poe:
return False
def load_json_file(self,filename:str):
# Loads json file
with open(filename) as f:
datajson = json.load(f)
return datajson
def fetch(self,result:Any):
# Callback function that fetches data after an SQL command is run self.runcommand("SELECT * FROM names WHERE name LIKE 'bill'",self.fetch)
return result
def json_to_sql(self,datajson :Union[dict,list]):
# Converts JSON to SQL.
if type(datajson) == list:
columns = str(tuple(datajson[0].keys())).replace("'","")
values = str(tuple(tuple(data.values()) for data in datajson))[1:-1]
return columns,values
elif type(datajson) == dict:
columns = str(tuple(datajson.keys())).replace("'","")
values = str(tuple(datajson.values())).replace("'","")
return columns,values
else:
print("JSON is invalid data shape.")
return None,None
def run_command(self,sqlcommand : str = None,result_function : Callable =None,datatuple : tuple =None,filename :str = None,verbose:int=1):
# Executes SQL Command or takes SQL file as input.
if verbose == 1:
if self.connection.is_connected():
db_Info = self.connection.get_server_info()
print("Connected to MySQL Server version ", db_Info)
if sqlcommand == None and filename == None:
print("Please input an SQL command or SQL filename.")
else:
if filename != None:
with open(filename) as f:
sqlcommand = f.read()
try:
with self.connection.cursor() as cursor:
#print(datatuple)
cursor.execute(sqlcommand,datatuple,multi=True)
result = cursor.fetchall()
if result_function != None:
new_result = result_function(result)
elif result_function == None:
new_result = None
#self.connection.commit()
if verbose == 1:
print("SQL command executed.")
return new_result
except Exception as poe:
print(f"{type(poe)} - {poe}")
def sql_to_json(self,table,sqldata :tuple):
# Convert SQL tuple to json
columnsinfo = self.run_command(f"DESCRIBE {table}",self.fetch)
columns = [col[0] for col in columnsinfo]
#print(sqldata)
final_json = []
for data in sqldata:
record = {}
for ind in range(len(data)):
record.update({data[ind]: columns[ind]} )
final_json.append(record)
return {table:final_json}
@staticmethod
def convert_to_blob(filename :str):
# Convert digital data to binary format
with open(filename, 'rb') as file:
blobData = file.read()
return blobData
@staticmethod
def start_docker_db(verbose=1):
# Start the docker database
# Run the database commands - "docker exec -it mysql mysql -p"
# https://mothishdeenadayalan.medium.com/containerizing-a-python-app-mysql-python-docker-1ce64e444ed9
dockercommand = 'docker run --name mysql -p 3306:3306 -v mysql_volume:/var/lib/mysql/ -d -e "MYSQL_ROOT_PASSWORD=temp123" mysql'
process = subprocess.Popen(dockercommand.split(),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = process.communicate()
if stderr != b"" and verbose == 1:
print(stderr)
elif stderr == b"" and verbose == 1:
print(stdout)
time.sleep(2)
return stdout,stderr
if __name__ == "__main__":
#CaesarSQL.start_docker_db()
caesarsql = CaesarSQL()
caesarsql.run_command(filename="customer.sql")
print("SQL file has been executed.")
|