pymysql數(shù)據(jù)庫連接
Note：PyMySQL 是在 Python3.x 版本中用于連接 MySQL 服務(wù)器的一個庫芥备，Python2中則使用mysqldb夯接。
- 知識點
1.連接數(shù)據(jù)
2.創(chuàng)建游標(biāo)
3.查詢數(shù)據(jù)庫
4.更改數(shù)據(jù)庫、提交
5.關(guān)閉連接
pymysql查詢數(shù)據(jù)庫
樣例：pymysqlOpreation.p
"""
創(chuàng)建數(shù)據(jù)庫連接
"""
class pymysqlopreations:
def __init__(self, host, port, user, password, db, charset='utf8', link_type=0):
"""
link_type = 0 時，數(shù)據(jù)庫返回結(jié)果為元組模式
當(dāng)link_type !=0時场梆，數(shù)據(jù)庫返回結(jié)果為字典模式
:param host:
:param port:
:param user:
:param password:
:param db:
:param charset:
:param link_type:
"""
self.host = host
self.port = port
self.user = user
self.password = password
self.db = db
self.charset = charset
try:
self.conn = pymysql.connect(host=self.host, port=self.port, user=self.user, password=self.password,
db=self.db, charset=self.charset)
if link_type == 1:
self.cur = self.conn.cursor(pymysql.cursors.DictCursor)
else:
self.cur = self.conn.cursor()
except Exception:
mylogger.info('Mysql數(shù)據(jù)庫連接失斒洹！')
def select_one(self, params):
"""
查詢一條數(shù)據(jù)
:param params:
:return:
"""
try:
self.cur.execute(params)
result = self.cur.fetchone()
result = {'code': '0000', 'message': '執(zhí)行單條查詢操作成功', 'data': result}
except:
result = {'code': '9999', 'message': '執(zhí)行單條查詢異常', 'data': []}
mylogger.info('Mysql查詢失敾蛴汀寞忿！')
return result
def select_all(self, params):
"""
查詢所有內(nèi)容
:param params:
:return:
"""
try:
row = self.cur.execute(params)
if row > 0:
self.cur.scroll(0, mode='absolute')
results = self.cur.fetchall()
result = {'code': '0000', 'message': '執(zhí)行查詢所有數(shù)據(jù)成功', 'data': results}
else:
result = {'code': '0000', 'message': '執(zhí)行查詢所有數(shù)據(jù)成功', 'data': []}
except:
result = {'code': '9999', 'message': '執(zhí)行查詢所有數(shù)據(jù)異常', 'data': []}
mylogger.info('Mysql查詢多條數(shù)據(jù)失敗顶岸！')
return result
def opt_one(self, params):
"""
操作單條數(shù)據(jù)
:param params:
:return:
"""
try:
row = self.cur.execute(params)
self.conn.commit()
result = {'code': '0000', 'message': '執(zhí)行單條數(shù)據(jù)成功', 'data': int(row)}
except:
result = {'code': '9999', 'message': '執(zhí)行單條數(shù)據(jù)失敗', 'data': []}
mylogger.info('Mysql操作單條數(shù)據(jù)失敗')
return result
def opt_many(self, stmt, data):
"""
多條數(shù)據(jù)
:param stmt:
:param data:
:return:
"""
try:
row = self.cur.executemany(stmt, data)
self.conn.commit()
result = {'code': '0000', 'message': '插入多條數(shù)據(jù)成功', 'data': int(row)}
except:
result = {'code': '9999', 'message': '插入多條數(shù)據(jù)失敗', 'data': []}
mylogger.info('Mysql 插入多條數(shù)據(jù)失敗')
return result
def __del__(self):
if self.cur != None:
self.cur.close()
if self.conn != None:
self.conn.close()
if __name__ == "__main__":
conn = pymysqlopreations(host='127.0.0.1', port=3306, user='root', password='root', db='insurance',
link_type=1)
sql = " SELECT * FROM `interface_detail`"
resutl = conn.select_one(sql)
print(resutl)
sql_demo = """ INSERT INTO `interface_detail`(NAME,detail,param,check_value,exe_env,LEVEL,require_login) VALUES('xx','xx','xx','id=10',0,0,0)"""
roe = conn.opt_one(sql_demo)
print(roe)
results = conn.select_all(sql)
print(results)
updatestmt = ''' update `interface_detail` set LEVEL = %s where interface_id =%s'''
data = [(5,18)]
conn.opt_many(updatestmt,data)
results = conn.select_all(sql)
print(results)
