pymysql操作mysql封装
时间：2021-07-01 10:21:17
帮助过：5人阅读
import pymysql
2 from lib.logger
import logger
3 from warnings
import filterwarnings
4 filterwarnings(
"ignore",category=
pymysql.Warning) # 忽略mysql警告信息
5
6 class MysqlDb():
7
8 logger =
logger
9
10 def __init__(self, host, port, user, passwd):
11 # 建立数据库连接
12 self.conn =
pymysql.connect(
13 host=
host,
14 port=
port,
15 user=
user,
16 passwd=
passwd,
17 charset=
‘utf8‘
18 )
19 # 通过 cursor() 创建游标对象，并让查询结果以字典格式输出
20 self.cur = self.conn.cursor(cursor=
pymysql.cursors.DictCursor)
21
22 def __del__(self):
# 对象资源被释放时触发，在对象即将被删除时的最后操作
23 # 关闭游标
24 self.cur.close()
25 # 关闭数据库连接
26 self.conn.close()
27
28 def select_db(self,db_name):
29 """
30 选择数据库
31 :return:
32 """
33 try:
34 self.conn.select_db(db_name)
35 self.logger.logger.info(
"选择 {0} 执行成功".format(db_name))
36 except Exception as e:
37 self.logger.logger.exception(
"操作出现错误：{0}".format(e))
38
39 def query_db(self, sql,state=
"all"):
40 """
41 查询
42 :param sql:sql语句
43 :param state:all查询所有，outhor查询单条
44 :return:
45 """
46 # 使用 execute() 执行sql
47 self.logger.logger.info(sql)
48 self.cur.execute(sql)
49 if state ==
"all":
50 # 使用 fetchall() 获取查询结果
51 data =
self.cur.fetchall()
52 else:
53 data =
self.cur.fetchone()
54 return data
55
56 def execute_db(self, sql):
57 """更新/插入/删除"""
58 try:
59 # 使用 execute() 执行sql
60 self.cur.execute(sql)
61 # 提交事务
62 self.conn.commit()
63 self.logger.logger.info(
"{0} 执行成功".format(sql))
64 except Exception as e:
65 self.logger.logger.exception(
"操作出现错误：{}".format(e))
66 # 回滚所有更改
67 self.conn.rollback()
68
69 if __name__ ==
‘__main__‘:
70 pass
pymysql操作mysql封装
标签：获取 hal from utf8 执行 div rollback 对象 事务
