84 lines
2.5 KiB
Python
84 lines
2.5 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.conn.autocommit(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)
|
|
|
|
def execute_write(self, sql, values=(), expected_rows=None):
|
|
"""
|
|
Execute an INSERT, UPDATE, or DELETE statement.
|
|
expected_rows:
|
|
None = do not check the number of affected rows
|
|
1 = expect exactly one affected row
|
|
0 = expect no affected rows
|
|
"""
|
|
try:
|
|
self.cur.execute(sql, values)
|
|
except MySQLdb.Error as e:
|
|
self.rollback()
|
|
self.error = e
|
|
return False
|
|
if (expected_rows is not None and self.cur.rowcount != expected_rows):
|
|
self.rollback()
|
|
self.error = RuntimeError(
|
|
"Expected {} affected rows, got {}".format(
|
|
expected_rows,
|
|
self.cur.rowcount
|
|
)
|
|
)
|
|
return False
|
|
self.error = None
|
|
return True
|