MySql公共方法
#!/usr/bin/env python
# coding:utf-8
import pymysql
from utils.logger import log
from config.env.pathconfig import ENV_CONFIG_PATH, API_YML_PATH
from utils.common import load_config, load_yml
import datetime
import decimal
from utils.DB import DB
db = DB()
sql_data = load_yml(API_YML_PATH)
class MySql(object):
def __init__(self, env=''):
db_info = load_config(ENV_CONFIG_PATH)[env]
# 数据库初始化
self.host = db_info['host']
self.port = int(db_info['port'])
self.user = db_info['user']
self.passwd = db_info['passwd']
self.database = db_info['database']
self.con = None
self.active = False
def _start(self):
'''
connect database if it's not active
:return:
'''
if not self.active:
self.connect()
def execute_sql(self, sql):
user_list = db.query_all(sql_data['delete_db_user'])
if 'delete' in sql:
for i in user_list:
data_base = i['database_list'].split(',')
for j in data_base:
if j in sql:
log.info('替换用户名')
self.user = i['username']
self.passwd = i['passwd']
self._start() # connect db if it's not active
self.con.automommit = True
log.info('execute sql {0}'.format(sql))
if sql.startswith('select') or sql.startswith('SELECT'):
result = self.query_all(sql)
return result
elif sql.startswith('drop') or sql.startswith('DROP'):
return None, '禁止使用drop命令'
elif sql.startswith('create') or sql.startswith('CREATE'):
return None, '禁止使用create命令'
else:
result = self.change_datas(sql)
return result
# self.closes()
def fetch_data(self, sql, limited_row):
data = []
cursors = self.con.cursor()
log.info('execute query all sql: {0}'.format(sql))
try:
cursors.execute(sql)
if not limited_row:
resu = cursors.fetchone()
data.append(resu)
else:
data = cursors.fetchmany(int(limited_row))
cursors.close()
log.info('query all result: {0}'.format(data))
return data, 'sql操作执行成功!'
except Exception as e:
log.error('query failed with error: {0}'.format(e))
return None, 'sql error! {}'.format(e)
def query_all(self, sql):
result = []
cursors = self.con.cursor()
log.info('execute query all sql: {0}'.format(sql))
try:
cursors.execute(sql)
data = cursors.fetchall()
for i in data:
for k, j in i.items():
if isinstance(j, datetime.datetime):
i[k] = datetime.datetime.strftime(
j, '%Y-%m-%d %H:%M:%S')
if isinstance(j, decimal.Decimal):
i[k] = str(j)
result.append(i)
cursors.close()
log.info('query all result: {0}'.format(data))
return result, 'sql操作执行成功!'
except Exception as e:
log.error('query failed with error: {0}'.format(e))
return None, 'sql error! {}'.format(e)
def change_datas(self, sql):
# 增，删，改
cursors = self.con.cursor()
log.info('update database with sql: {0}'.format(sql))
try:
effect_row = cursors.execute(sql)
self.con.commit() # commit update
cursors.close() # close cursor
log.info('inert into result: {0}'.format(effect_row))
return effect_row, 'sql操作执行成功!'
except Exception as e:
log.error('inert into failed with error: {0}'.format(e))
self.con.rollback() # rollback if update failed
return None, 'sql error! {}'.format(e)
def execute_batch_import_sql(self, sql, values):
try:
self._start()
cursors = self.con.cursor()
result = cursors.executemany(sql, values)
cursors.close()
self.con.commit()
return result
except Exception as err:
log.error('import failed with error: {0}'.format(err))
return None
def connect(self):
# 连接数据库
try:
connection = pymysql.connect(
host=self.host,
port=int(self.port),
user=self.user,
passwd=self.passwd,
database=self.database,
cursorclass=pymysql.cursors.DictCursor,
charset='utf8',
use_unicode=True)
log.info('{0} :database is connecting successful'.format(
self.host))
self.con = connection
except Exception as e:
log.error('{0} :database is connecting failed : {1}'.format(
self.host, e))
def closes(self):
# 关闭数据库连接
try:
self.active = False
self.con.close()
log.info('database is closed: {0}'.format(self.host))
except Exception as e:
self.active = False
log.error('database closed failed with error: {0}'.format(e))
raise 'server is error {0}'.format(e)