#!/usr/bin/env python
import mysql.connector # for connectivity to MySQL database
[docs]class DbMan:
"""
The dbMan class represents a single database accessed across a specific MySQL connector. It is initialised by passing a set of credentials
for opening a connection to a MySQL server; these credentials are passed as a dictionary which defines the values for the 'user', 'password',
'host', 'port' and 'database' variables to be used to access an existing database on a MySQL server and the value of the 'raise_on_warnings' flag
to be used over the connection to that database.
The class supports new table, query and update functions which operate on the database, and 'serverInfo' and 'close' functions which operate on the connection.
Work ongoing!
"""
def __init__(self, credentials, dB):
self.host = credentials["host"]
self.database = dB
self.status = "Connected"
self.statusDetail = "Connected to the " + self.database + " database, hosted on " + self.host
try:
credentials["database"] = dB
self.conn = mysql.connector.connect(**credentials)
if self.conn.is_connected():
self.cur = self.conn.cursor()
print(self.statusDetail)
except mysql.connector.Error as error:
self.status = "Failed"
self.statusDetail = "Failed to connect to {} on {}: {}".format(self.database, self.host, error)
print(self.statusDetail)
[docs] def listTables(self):
genericCode = "SHOW TABLES"
return self.cur.execute(genericCode)
[docs] def createTable(self, tableName, tableColumns):
genericCode = "CREATE TABLE {} ({})"
sqlCode = genericCode.format(tableName, tableColumns)
self.cur.execute(sqlCode)
[docs] def dropTable(self, table):
genericCode = "DROP TABLE IF EXISTS {}"
sqlCode = genericCode.format(table)
self.cur.execute(sqlCode)
[docs] def queryTable(self, sqlCode):
self.cur.execute(sqlCode)
return self.cur.fetchall()
[docs] def updateTable(self, sqlCode, data):
self.cur.executemany(sqlCode, data)
self.conn.commit()
self.status = str(self.cur.rowcount) + " records inserted successfully"
return self.status
[docs] def singleupdateTable(self, sqlCode, data):
self.cur.execute(sqlCode, data)
self.conn.commit()
self.status = str(self.cur.rowcount) + " records inserted successfully"
return self.status
[docs] def close(self):
if (self.conn.is_connected()):
self.cur.close()
self.conn.close()
print("Connection to {} on {} is closed".format(self.database, self.host))
else: print("No connection to close!")
[docs]class SvrMan:
"""
The SvrMan class represents a single server accessed across a specific MySQL connector. It is initialised by passing a set of credentials
for opening a connection to a MySQL server; these credentials are passed as a dictionary which defines the values for the 'user', 'password',
'host', and 'port' variables to be used to access an existing database on a MySQL server and the value of the 'raise_on_warnings' flag
to be used over the connection to that database.
The class supports new table, query and update functions which operate on the database, and 'serverInfo' and 'close' functions which operate on the connection.
Work ongoing!
..todo: Really DbmMan should 'inherit from' SvrMan. Need to work out how inheritance works in python.
"""
def __init__(self, credentials):
self.host = credentials["host"]
self.status = "Connected"
self.statusDetail = "Connected to " + self.host
try:
self.conn = mysql.connector.connect(**credentials)
if self.conn.is_connected():
self.cur = self.conn.cursor()
except mysql.connector.Error as error:
self.status = "Failed"
self.statusDetail = "Failed to connect to {}: {}".format(self.host, error)
print(self.statusDetail)
[docs] def listDatabases(self):
genericCode = "SHOW DATABASES"
try:
if self.conn.is_connected(): self.cur.execute(genericCode)
except mysql.connector.Error as error:
self.status = "Failed"
self.statusDetail = "Failed to create database '{}': {}.".format(dB, error)
finally:
return self.cur.fetchall()
[docs] def dropDatabase(self, dB):
genericCode = "DROP DATABASE IF EXISTS {}"
sqlCode = genericCode.format(dB)
self.cur.execute(sqlCode)
[docs] def createDatabase(self, dB):
genericCode = "CREATE DATABASE {}"
sqlCode = genericCode.format(dB)
try:
if self.conn.is_connected(): self.cur.execute(sqlCode)
except mysql.connector.Error as error:
self.status = "Failed"
self.statusDetail = "Failed to create database '{}': {}.".format(dB, error)
finally:
return self.status
[docs] def serverInfo(self):
return self.conn.get_server_info()
[docs] def close(self):
if (self.conn.is_connected()):
self.cur.close()
self.conn.close()
print("Connection to {} is closed".format(self.host))
else: print("No connection to close!")
[docs]class table:
""" A Table class object contains the column definitions of the table that it has been instantiated to represent. The column structure has to be provided as a tuple of tuples, each individual tuple being constituted of three strings that specify the column name, datatype and datatype again expressed as a % term (mostly %s)
A table object includes a createSQL() function which returns the SQL code to create the table, and insertSQL which returns the SQL code to insert
:param content: Set of options to be presented to the user, defaults to (no default option)
:type content: tuple of tuples of strs"""
def __init__(self, content):
self.columns = content
[docs] def createSQL(self):
SQLtext1 = ""
for p in self.columns:
SQLtext1 = SQLtext1 + '{} {}, '.format(p[0], p[1])
result = (SQLtext1.rstrip(", "))
return result
[docs] def insertSQL(self):
SQLtext1 = ""
SQLtext2 = ""
for p in self.columns:
SQLtext1 = SQLtext1 + '{}, '.format(p[0])
SQLtext2 = SQLtext2 + '{}, '.format(p[2])
result = (SQLtext1.rstrip(", "), SQLtext2.rstrip(", "))
return result