class DBTool(object):
conn = None
cursor = None
def __init__(self, conn_dict):
self.conn = pymysql.connect(host=source['host'],
port=source['port'],
user=source['user'],
passwd=source['passwd'])
self.cursor = self.conn.cursor()
self.cursor.execute("set character_set_client = 'utf8';")
self.cursor.execute("set character_set_connection = 'utf8';")
self.cursor.execute("set character_set_results = 'utf8';")
self.cursor.execute("set session group_concat_max_len = 102400;")
def execute_query(self, sql_string):
try:
cursor = self.cursor
cursor.execute(sql_string)
list = cursor.fetchall()
cursor.close()
self.conn.close()
return list
except pymysql.Error as e:
print("mysql execute error:", e)
raise
def execute_noquery(self, sql_string):
try:
cursor = self.cursor
cursor.execute(sql_string)
self.conn.commit()
self.cursor.close()
self.conn.close()
except pymysql.Error as e:
print("mysql execute error:", e)
raise
|