前言

在 python 中我们想要操作 mysql 需要用到 pymysql 这个库。

日常写个 爬虫 将爬下来的数据，往数据库中存。或者是批量插入数据。

都要用到 pymysql 这个第三方库。

但是 pymysql 的操作又比较复杂！又是连接数据库又是创建游标对象的太麻烦了！

为此我针对 pymysql 进行了简单的进行了一下封装。

效果示例

import os
# 导入自定义模块 PyMysqlModel
from PyMysqlModel.Model import Model
student = Model(database="pymysqlmodel",user="root",password=123)
# 表名
table_name = "student_tb"
# 表字段
student_table_fields = [
"name varchar (20)",
# 一个字符串为一个字段
"age int",
"gender enum('男','女')",
"phone varchar(11)",
]
# 连接指定表
# 不存在创建
student.link_table(table_name,student_table_fields)
names = "张三"
ages = 18
genders = 1
phones = "17613355211"
# 添加数据
# flag = student.create(name=names,age=ages,gender=genders,phone=phones)
# print(flag)
# 获取所有数据
data = student.all() # 获取所有信息
for i in data:
print(i)
# pk = 2
# names = "李四"
# ages = 21
# genders = 2
# phones = "17613355211"
# 修改数据
# flag = student.update(id=pk,name=names, age=ages, gender=genders, phone=phones)
# print(flag)
# 删除数据
pk = 3
names = "张三"
# flag = student.delete(id=pk)
# flag = student.delete(id=pk,name=names)
# print(flag)
# 查询
# 根据表字段进行过滤查询
ages = 18
data = student.filter(age=ages)
print("查询结果")
for i in data:
print(i)
# 获取单条数据
names = "李四"
data = student.get(name=names)
print("查询单挑结果")
print(data)

该模块已发布到 PyPi 上。
可直接 pip 安装

pip 安装

pip install PyMysqlModel

pymsql封装

源码如下：

import pymysql
# mysql操作
class Model():
def __init__(self,database=None,user=None,password=None,host="localhost",port=3306,charset="utf8"):
if not all([database, user, password, host, port, charset]):
try:
from setting import DATABASES
except Exception:
raise ValueError("setting DATABASES 不存在！")
self.pymysql_con = pymysql.connect(**DATABASES)
else:
self.pymysql_con = pymysql.connect(database=database, user=user, password=str(password), host=host,port=port, charset=charset)
self.mysql = self.pymysql_con.cursor() # 创建游标对象
self.table_name = None
# 表名
self.field = ["id",]
# 表字段
# 获取数据库内所有的表
def show_tables(self):
sql = f"""show tables;"""
self.mysql.execute(sql)
data = self.mysql.fetchall()
table_list = []
for i in data:
# 元组转换为列表
table_list.append(i[0])
return table_list # 返回当前数据库内所有的表
# 连接表 接收一个字符串类型的表名
def link_table(self, table_name=None, field=None):
self.table_name = table_name
# 将表名赋值给实例属性
for i in field:
field_name = i.split()[0]
# 获取所有的字段名，不含字段类型
self.field.append(field_name)
# 获取该表的所有的字段名
table_list = self.show_tables()
# 获取数据库里所有的表
for i in table_list:
if self.table_name == i:
# 判断该表是否已存在
print(f"——连接成功：{self.table_name}——")
return True # 该表已存在！直接返回
create_field = ",".join(field)
# 将所有的字段与字段类型以 “ , ” 拼接
sql = f"""
create table {self.table_name}(
id int primary key auto_increment,
{create_field}
);
"""
self.mysql.execute(sql)
print(f"——创建并连接成功：{self.table_name}——")
return True
# 查询数据
def all(self,*args):
if len(args) != 0:
field = args
else:
field = self.field
for i in field:
assert i in self.field,f"{i} 该字段不存在！"
select_field = ",".join(field)
sql = f"""select {select_field} from {self.table_name};""" # 根据表名直接查询
self.mysql.execute(sql)
data = self.mysql.fetchall()
result = []
for i in data:
# 进行数据转换
temp = {}
for k,j in enumerate(field):# 每行数据 组成字典类型
temp[j] = i[k]
result.append(temp)
return result # 最终返回查询集
# 添加数据
接收所有的命名参数 即根据字段名传入对应数据
def create(self,**kwargs):
try:
value_list = []
for i in self.field:
value = kwargs.get(i)
if value != None:
value_list.append(f"'{value}'")
else:
value_list.append(f"null")
create_sql = ",".join(value_list)
# id 字段为null ，默认自增
sql = f"""
insert into {self.table_name} values
({create_sql});
"""
self.mysql.execute(sql)
except Exception as err:
self.pymysql_con.rollback()
print("错误信息：", err)
return False
else:
self.pymysql_con.commit()
return True
# 修改数据
接收所有的命名参数 即根据字段名传入对应数据
def update(self,**kwargs):
try:
pk = kwargs.get("id")
assert pk,"缺少 id 字段"
field = []
temp = kwargs.keys()
for i in self.field:
if i in temp:
field.append(i)
value_list = []
for i in field:
value = kwargs.get(i)
if value != None:
value_list.append(f"{i}='{value}'")
else:
value_list.append(f"{i}=null")
update_sql = ",".join(value_list)
sql = f"""
update {self.table_name} set {update_sql} where id = {pk};
"""
self.mysql.execute(sql)
except Exception as err:
self.pymysql_con.rollback()
print("错误信息：", err)
return False
else:
self.pymysql_con.commit()
return True
# 删除数据的方法
接收所有的命名参数 即根据字段名传入对应数据
def delete(self,**kwargs):
try:
field = []
temp = kwargs.keys()
for i in self.field:
if i in temp:
field.append(i)
value_list = []
for i in field:
value = kwargs.get(i)
if value != None:
value_list.append(f"{i}='{value}'")
else:
value_list.append(f"{i}=null")
delete_sql = " and ".join(value_list)
# 先查询满足条件的数据个数
sql = f"""
select COUNT(id) from {self.table_name} where {delete_sql};
"""
self.mysql.execute(sql)
data = self.mysql.fetchall()
delete_sum = data[0][0]
# 删除数据
sql = f"""
delete from {self.table_name} where {delete_sql};
"""
self.mysql.execute(sql)
except Exception as err:
self.pymysql_con.rollback()
print("错误信息：", err)
return False
else:
self.pymysql_con.commit()
return delete_sum
# 过滤查询数据
def filter(self, *args, **kwargs):
try:
field = []
temp = kwargs.keys()
for i in self.field:
if i in temp:
field.append(i)
value_list = []
for i in field:
value = kwargs.get(i)
if value != None:
value_list.append(f"{i}='{value}'")
else:
value_list.append(f"{i}=null")
filter_sql = " and ".join(value_list)
# 结果字段
if len(args) != 0:
result_field = args
else:
result_field = self.field
select_field = ",".join(result_field)
for i in result_field:
assert i in self.field, f"{i} 该字段不存在！"
sql = f"""
select {select_field} from {self.table_name} where {filter_sql};
"""
self.mysql.execute(sql)
except Exception as e:
print("错误信息：", e)
else:
data = self.mysql.fetchall()
result = []
for i in data:
temp = {}
for k,j in enumerate(result_field):
temp[j] = i[k]
result.append(temp)
# 返回查询集
return result
# 过滤获取数据
返回第一个数据
def get(self,**kwargs):
try:
field = []
temp = kwargs.keys()
for i in self.field:
if i in temp:
field.append(i)
value_list = []
for i in field:
value = kwargs.get(i)
if value != None:
value_list.append(f"{i}='{value}'")
else:
value_list.append(f"{i}=null")
get_sql = " and ".join(value_list)
sql = f"""
select * from {self.table_name} where {get_sql};
"""
self.mysql.execute(sql)
except Exception as err:
print("错误信息：", err)
else:
data = self.mysql.fetchall()
data = data[0]
result = {}
for k, j in enumerate(self.field):
result[j] = data[k]
return result

总结

如果觉得还不错，那就点个赞吧！

欢迎大家使用！如果发现有什么bug欢迎在评论区留言！

个人地址

最后

以上就是简单猫咪为你收集整理的基于pymysql高阶封装，PyMysqlModel前言效果示例pip 安装pymsql封装总结的全部内容，希望文章能够帮你解决基于pymysql高阶封装，PyMysqlModel前言效果示例pip 安装pymsql封装总结所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。