57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import MySQLdb
|
|
|
|
class Database():
|
|
|
|
def __init__(self, hostname, dbname):
|
|
self.hostname = hostname
|
|
self.schema = dbname
|
|
self.conn = None
|
|
self.cur = None
|
|
self.dcur = None
|
|
self.error = None
|
|
|
|
def connect(self, dbuser, dbpass):
|
|
if self.conn:
|
|
return True
|
|
MySQLdb.paramstyle = 'pyformat'
|
|
try:
|
|
self.conn = MySQLdb.connect(host=self.hostname, db=self.schema,
|
|
user=dbuser, passwd=dbpass, charset='utf8')
|
|
except MySQLdb.Error as e:
|
|
self.error = e
|
|
return False
|
|
self.cur = self.conn.cursor()
|
|
self.dcur = self.conn.cursor(MySQLdb.cursors.DictCursor)
|
|
return True
|
|
|
|
def connect_local(self):
|
|
# connection via local unix socket
|
|
if self.conn:
|
|
return True
|
|
MySQLdb.paramstyle = 'pyformat'
|
|
try:
|
|
self.conn = MySQLdb.connect(host='localhost', db=self.schema,
|
|
unix_socket='/run/mysqld/mysqld.sock', charset='utf8')
|
|
except MySQLdb.Error as e:
|
|
self.error = e
|
|
return False
|
|
self.cur = self.conn.cursor()
|
|
self.dcur = self.conn.cursor(MySQLdb.cursors.DictCursor)
|
|
return True
|
|
|
|
def disconnect(self):
|
|
self.cur.close()
|
|
self.dcur.close()
|
|
self.conn.close()
|
|
|
|
def is_connected(self):
|
|
return self.conn.open
|
|
|
|
def last_error(self):
|
|
if not self.error:
|
|
return(0, '')
|
|
code = self.error.args[0]
|
|
text = self.error.args[1]
|
|
self.error = None
|
|
return(code, text)
|