Python操作MySQL工具类

在上一篇文章中&＃xff0c;简单对Python利用pymysql操作MySQL进行了封装&＃xff0c;实现了多种功能的复用。接下来&＃xff0c;在参阅pymysql的API后&＃xff0c;对Python进行了更加详细的封装&＃xff0c;实现了Python操作MySQL的工具类MySQLDBHelper&＃xff0c;其代码主要如下&＃xff1a;

MySQLDBHelper工具类

import pymysqlimport reclass MySQLDBHelper(object): """ 定义构造方法&＃xff0c;用于初始化数据库连接 """ def __init__(self, config): self.host &＃61; config[&＃39;host&＃39;] self.user &＃61; config[&＃39;user&＃39;] self.password &＃61; config[&＃39;password&＃39;] self.port &＃61; config[&＃39;port&＃39;] self.connection &＃61; None self.cursor &＃61; None try: self.connection &＃61; pymysql.connect(**config) self.connection.autocommit(1) # 调用cursor()方法创建一个用于操作的游标 self.cursor &＃61; self.connection.cursor() except: print("数据库连接失败&＃xff0c;请检查数据库配置项!") # 定义关闭数据库连接的方法 def close(self): if not self.connection: self.connection.close() else: print("数据库连接已经关闭!") # 创建数据库 def createDatabase(self, DB_NAME): """ 创建数据库 """ self.cursor.execute( "CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci" % DB_NAME) self.connection.select_db(DB_NAME) print(f"数据库{DB_NAME}创建成功!") # 选择数据库 def selectDatabase(self, DB_NAME): self.connection.select_db(DB_NAME) # 获取数据库的版本号 def getDatabaseVersion(self): self.cursor.execute(&＃39;SELECT VERSION()&＃39;) return self.queryOne() # 获取一行查询结果 def queryOne(self): return self.cursor.fetchone() # 创建数据库表 def createTable(self, table_name, attribute_dict, constraint): """ 创建数据库表 :param table_name: 表的名称 :param attribute_dict: 属性键值对 :param constraint: 主外键约束 """ # 判断要创建的表是否存在 if self.isExistTable(table_name): print(f"{table_name}已经存在!") return sql &＃61; &＃39;&＃39; sql_mid &＃61; &＃39;&＃96;id&＃96; bigint(11) auto_increment,&＃39; for attr, value in attribute_dict.items(): sql_mid &＃61; sql_mid &＃43; &＃39;&＃96;&＃39; &＃43; attr &＃43; &＃39;&＃96;&＃39; &＃43; &＃39; &＃39; &＃43; value &＃43; &＃39;,&＃39; sql &＃61; sql &＃43; &＃39;CREATE TABLE IF NOT EXISTS %s (&＃39; % table_name sql &＃61; sql &＃43; sql_mid sql &＃61; sql &＃43; constraint sql &＃61; sql &＃43; &＃39;) ENGINE&＃61;InnoDB DEFAULT CHARSET&＃61;utf8&＃39; print(f&＃39;创建表的语句是&＃xff1a;{sql}&＃39;) self.executeCommit(sql) def executeSql(self, sql&＃61;&＃39;&＃39;): """ 执行sql语句&＃xff0c;针对读操作返回结果集 """ try: self.cursor.execute(sql) records &＃61; self.cursor.fetchall() return records except pymysql.Error as e: print(f&＃39;SQL语句执行失败&＃xff01;ERROR({e.args[0]}): {e.args[1]}&＃39;) def executeCommit(self, sql&＃61;&＃39;&＃39;): """ 执行SQL语句&＃xff0c;针对更新、删除等事务操作&＃xff0c;失败时回滚 """ try: self.cursor.execute(sql) self.connection.commit() except pymysql.Error as e: self.connection.rollback() error &＃61; &＃39;SQL语句执行失败&＃xff01;ERROR (%s): %s&＃39; % (e.args[0], e.args[1]) print(error) return error def insert(self, table_name, params): """ """ key &＃61; [] value &＃61; [] for k, v in params.items(): key.append(k) if isinstance(v, str): value.append("&＃39;" &＃43; v &＃43; "&＃39;") else: value.append(v) attrs_sql &＃61; &＃39;(&＃39; &＃43; &＃39;,&＃39;.join(key) &＃43; &＃39;)&＃39; values_sql &＃61; &＃39; values(&＃39; &＃43; &＃39;,&＃39;.join(value) &＃43; &＃39;)&＃39; sql &＃61; &＃39;insert into %s&＃39; % table_name sql &＃61; sql &＃43; attrs_sql &＃43; values_sql print(f&＃39;插入SQL语句: {sql}&＃39;) self.executeCommit(sql) def select(self, table_name, cond_dict&＃61;&＃39;&＃39;, order&＃61;&＃39;&＃39;, fields&＃61;&＃39;*&＃39;): """ 查询数据 args: table_name: 表的名字 cond_dict: 查询条件 order: 排序条件 示例&＃xff1a; db.select(table) db.select(table, fields&＃61;[&＃39;user_name]) db.select(table, fields&＃61;[&＃39;user_name&＃39;,&＃39;income&＃39;]) """ cond_sql &＃61; &＃39; &＃39; if cond_dict !&＃61; &＃39;&＃39;: for k, v in cond_dict.items(): cond_sql &＃61; cond_sql &＃43; &＃39;&＃96;&＃39; &＃43; k &＃43; &＃39;&＃96;&＃39; &＃43; &＃39;&＃61;&＃39; &＃43; &＃39;"&＃39; &＃43; v &＃43; &＃39;"&＃39; &＃43; &＃39; and&＃39; cond_sql &＃61; cond_sql &＃43; &＃39; 1&＃61;1 &＃39; if fields &＃61;&＃61; &＃39;*&＃39;: sql &＃61; &＃39;select * from %s where &＃39; % table_name else: if isinstance(fields, list): fields &＃61; ",".join(fields) sql &＃61; &＃39;select %s from %s where &＃39; % (fields, table_name) else: print(&＃39;传入的字段不正确&＃xff0c;请以列表的形式传入字段信息!&＃39;) sql &＃61; sql &＃43; cond_sql &＃43; order print(f&＃39;查询语句: {sql}&＃39;) return self.executeSql(sql) def insertMany(self, table_name, attrs, values): """ 插入多条数据 args: table_name: 表的名字 attrs: 属性键 values: 属性值 示例: table_name &＃61;&＃39;mytable&＃39; key &＃61; ["id","name","age"] value &＃61; [[1,&＃39;张三&＃39;,20],[2,&＃39;李四&＃39;,25]] db.insertMany(table_name, key, value) """ values_sql &＃61; [&＃39;%s&＃39; for attr in attrs] attrs_sql &＃61; &＃39;(&＃39; &＃43; &＃39;,&＃39;.join(attrs) &＃43; &＃39;)&＃39; values_sql &＃61; &＃39; values(&＃39; &＃43; &＃39;,&＃39;.join(values_sql) &＃43; &＃39;)&＃39; sql &＃61; &＃39;insert into %s&＃39; % table_name sql &＃61; sql &＃43; attrs_sql &＃43; values_sql print(f&＃39;插入多条语句:{sql}&＃39;) try: for i in range(0, len(values), 20000): self.cursor.executemany(sql, values[i:i &＃43; 20000]) self.connection.commit() except pymysql.Error as e: self.connection.rollback() print(f&＃39;插入多条数据遇到错误! ERROR({e.args[0]}):{e.args[1]}&＃39;) def delete(self, table_name, cond_dict): """ 删除数据 args: table_name: 表的名字 cond_dict: 删除条件字典 示例&＃xff1a; params &＃61; {&＃39;name&＃39;: &＃39;风清扬&＃39;, &＃39;age&＃39;:38} db.delete(table_name, params) """ cond_sql &＃61; &＃39; &＃39; if cond_sql !&＃61; &＃39;&＃39;: for k, v in cond_dict.items(): if isinstance(v, str): v &＃61; "&＃39;" &＃43; v &＃43; "&＃39;" cond_sql &＃61; cond_sql &＃43; table_name &＃43; "." &＃43; k &＃43; &＃39;&＃61;&＃39; &＃43; v &＃43; &＃39; and &＃39; cond_sql &＃61; cond_sql &＃43; &＃39; 1&＃61;1 &＃39; sql &＃61; &＃39;delete from %s where %s &＃39; % (table_name, cond_sql) print(f&＃39;生成的sql语句&＃xff1a;{sql}&＃39;) return self.executeCommit(sql) def update(self, table_name, attrs_dict, cond_dict): """ args: table_name: 表的名字 attrs_dict: 更新属性键值对字典 cond_dict: 更新条件字典 example: params &＃61; {&＃39;name&＃39;: &＃39;风清扬&＃39;, &＃39;age&＃39;: 40} """ attrs_list &＃61; [] cond_sql &＃61; &＃39; &＃39; for k, v in attrs_dict.items(): attrs_list.append("&＃96;" &＃43; k &＃43; "&＃96;" &＃43; "&＃61;" &＃43; "&＃39;" &＃43; v &＃43; "&＃39;") attrs_sql &＃61; ",".join(attrs_list) print(f"attrs_sql: {attrs_sql}") if cond_dict !&＃61; &＃39;&＃39;: for k, v in cond_dict.items(): if isinstance(v, str): v &＃61; "&＃39;" &＃43; v &＃43; "&＃39;" cond_sql &＃61; cond_sql &＃43; "&＃96;" &＃43; table_name &＃43; "&＃96;." &＃43; "&＃96;" &＃43; k &＃43; "&＃96;" &＃43; &＃39;&＃61;&＃39; &＃43; v &＃43; &＃39; and &＃39; cond_sql &＃61; cond_sql &＃43; &＃39; 1&＃61;1 &＃39; sql &＃61; &＃39;update %s set %s where %s&＃39; % (table_name, attrs_sql, cond_sql) print(sql) return self.executeCommit(sql) def dropTable(self, table_name): """ 删除数据库表 args: table_name: 表名字 """ sql &＃61; &＃39;DROP TABLE %s&＃39; % table_name self.executeCommit(sql) def deleteTable(self, table_name): """ 清空表数据 """ sql &＃61; &＃39;DELETE FROM %s&＃39; % table_name print(sql) self.executeCommit(sql) def truncateTable(self, table_name): """ 清空表数据&＃xff0c;不写日志 """ sql &＃61; &＃39;TRUNCATE TABLE %s&＃39; % table_name print(sql) self.executeCommit(sql) def isExistTable(self, table_name): """ 判断表是否存在 """ sql &＃61; &＃39;select * from %s&＃39; % table_name ret &＃61; self.executeCommit(sql) if ret is None: return True else: if re.search("doesn&＃39;t exist", ret): return False else: return True

MySQLDBHelper工具类的测试

# 测试if __name__ &＃61;&＃61; "__main__": # 定义数据库访问参数 config &＃61; { &＃39;host&＃39;: &＃39;node05&＃39;, &＃39;user&＃39;: &＃39;root&＃39;, &＃39;password&＃39;: &＃39;Love88me&＃39;, &＃39;port&＃39;: 3306, &＃39;charset&＃39;: &＃39;utf8&＃39;, &＃39;cursorclass&＃39;: pymysql.cursors.DictCursor } # 初始化打开数据库连接 db &＃61; MySQLDBHelper(config) # 打印数据库版本 print(db.getDatabaseVersion()) # 测试创建数据库 print("&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;测试创建数据库&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;") DB_NAME &＃61; input(&＃39;输入要创建的数据库名称&＃xff1a;&＃39;) db.createDatabase(DB_NAME) # 测试选择数据库 print("&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;测试选择数据库&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;") db.selectDatabase(DB_NAME) # 测试创建表 print("&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;测试创建数据表&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;") table_name &＃61; input(&＃39;请输入要创建的数据表名&＃xff1a;&＃39;) &＃39;&＃39;&＃39; 例如 create table &＃96;location&＃96; ( &＃96;id&＃96; bigint(11) not null auto_increment, &＃96;name&＃96; varchar(30) not null, &＃96;province&＃96; varchar(20), &＃96;city&＃96; varchar(50) ) engine&＃61;InnoDB default charset&＃61;utf8; &＃39;&＃39;&＃39; attrs_dict &＃61; { &＃39;name&＃39;: &＃39;varchar(30) not null&＃39;, &＃39;province&＃39;: &＃39;varchar(30) not null&＃39;, &＃39;city&＃39;: &＃39;varchar(50)&＃39; } constraint &＃61; &＃39;primary key(&＃96;id&＃96;)&＃39; db.createTable(table_name, attrs_dict, constraint) # 测试单条数据插入功能 params &＃61; {&＃39;name&＃39;: &＃39;北京市文化局&＃39;, &＃39;province&＃39;: &＃39;北京&＃39;, &＃39;city&＃39;: &＃39;丰台区&＃39;} db.insert(table_name, params) # 测试批量插入数据 insert_values &＃61; [[&＃39;江苏省无锡市文化局&＃39;,&＃39;江苏省&＃39;,&＃39;无锡市&＃39;], [&＃39;安徽省合肥市教育局&＃39;,&＃39;安徽省&＃39;,&＃39;合肥市&＃39;], [&＃39;浙江省杭州市旅游局&＃39;,&＃39;浙江省&＃39;,&＃39;杭州市&＃39;]] insert_attrs &＃61; [&＃39;name&＃39;,&＃39;province&＃39;,&＃39;city&＃39;] db.insertMany(table_name, insert_attrs, insert_values) # 测试数据查询 print(db.select(table_name, fields&＃61;[&＃39;id&＃39;,&＃39;name&＃39;])) print(db.select(table_name,cond_dict&＃61;{&＃39;province&＃39;:&＃39;江苏省&＃39;}, fields&＃61;[&＃39;name&＃39;,&＃39;province&＃39;,&＃39;city&＃39;])) print(db.select(table_name, fields&＃61;[&＃39;id&＃39;,&＃39;name&＃39;, &＃39;province&＃39;, &＃39;city&＃39;], order&＃61;&＃39;order by id desc&＃39;)) # 测试更新数据 update_params &＃61; {&＃39;name&＃39;:&＃39;江苏省无锡市苏南国际鸡厂&＃39;} update_cond_dict &＃61; {&＃39;province&＃39;:&＃39;江苏省&＃39;} db.update(table_name, update_params, update_cond_dict) # 测试删除数据 delete_params &＃61; {&＃39;province&＃39;:&＃39;安徽省&＃39;} db.delete(table_name, delete_params) # 测试删除表数据 db.deleteTable(table_name) # 测试删除表 db.dropTable(table_name)

操作的中间结果