- 前言
- pymysql运行逻辑
- pymsql封装
- 自定义模型层
- 定义 “ 模型类 ” 功能
- 模型层实现
- 视图使用
- 总结
前言
这两天刚学flask，还没用flask自带的模型层，让用pymysql实现对数据库的增删改查。
为此我又专门去复习了pymysql的使用方法等等。
总结的一点就是：真 · 麻烦。
这要实现增删改查，视图代码可能就几行。
其它的全是pymysql的操作方法，全篇代码占幅70%多。
害~
烦死了！
对于有强迫症的我来说，让我每个视图都写一遍重复的没意义的代码可真难受！
更何况可读性非常差！这才是致命的！
pymysql运行逻辑
其实就是把原生的sql语句，给pysmql让它执行就ok了。
这样的话我们直接在sql语句上下功夫就行了。
pymsql封装首先新建一个Msql文件。（注意不要与pymysql包名一致！）
先将pymysql进行第一次封装。
代码如下：
import pymysql
# mysql操作
class Mysql(): # 实例对象时，传入数据库配置
def __init__(self,host=None,port=None,user=None,password=None,database=None,charset=None):
try:
# 连接数据库
self.pymysql_con = pymysql.connect(host=host,port=port,user=user,password=password,database=database,charset=charset)
self.mysql = self.pymysql_con.cursor() # 创建游标对象
except Exception as e:
print("连接数据库失败！")
print("错误信息：",e)
else:
print("连接数据库成功！")
# 创建表
def create_table(self):
try:
sql = f"""
create table {self.table}(
id int primary key auto_increment,
name varchar (20),
age int ,
sex enum("男","女")
)
"""
self.mysql.execute(sql) # 执行sql语句
except Exception as e:
print("创建表失败！")
print("错误信息：", e)
else:
print("创建表成功！")
这只是一个简单的封装，在实例对象时连接数据库。
然后需要什么方法就写什么方法就可以，大体框架也就出来了。
不过这还非常简陋，sql语句也是死的，我们需要活的sql语句。
因为每个表的表结构都不相同！
所以现在的复用性非常差！
下面我们需要再次进行封装完善！
自定义模型层
因为我之前用的是django，所以我要实现的效果向django的模型类功能看齐。
至少要达到可以创建模型类，实现基本的增删改查。
所以这里首先要实现 “ 定义模型类 ” 这个功能。也就是建表！
定义 “ 模型类 ” 功能定义模型类说白了也就是建表。
创建表则需要：表名、表字段。
所以我们在视图层实例对象时需要传入：表名、表字段。
视图层代码如下：
# 导入封装的mysql
from Mysql import Mysql
# 实例mysql对象
student = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# 实例一个对象，相当于创建一个模型类，可创建多个模型类
# teacher = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# goods = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# 表字段
student_table_fields = [
"name varchar (20)", # 一个字符串为一个表字段
"age int",
"gender enum('男','女')",
"phone varchar(11)",
]
# 调用创建表方法 第一个参数为表名字符串类型， 第二个参数为表字段,类型为：列表、元组
student.create_table("student",student_table_fields)
这是我们要实现的效果，调用方法传入表名与表字段列表就可以创建该表了！
模型层代码如下：
再次封装！
# mysql操作
class Mysql():
def __init__(self,host=None,port=None,user=None,password=None,database=None,charset=None):
try:
self.table_name = None # 表名
self.field = ["id",] # 表字段
self.pymysql_con = pymysql.connect(host=host,port=port,user=user,password=password,database=database,charset=charset)
self.mysql = self.pymysql_con.cursor() # 创建游标对象
except Exception as e:
print("连接数据库失败！")
print("错误信息：",e)
else:
print("连接数据库成功！")
# 获取数据库内所有的表
def show_tables(self):
try:
sql = f"""show tables;"""
self.mysql.execute(sql)
except Exception as e:
print("查询表失败！")
print("错误信息：", e)
else:
data = self.mysql.fetchall()
table_list = []
for i in data: # 元组转换为列表
table_list.append(i[0])
return table_list # 返回当前数据库内所有的表
# 创建表 接收一个字符串类型的表名，和一个列表、元组类型的表字段
def create_table(self,table_name=None,field=None):
try:
self.table_name = table_name # 将表名赋值给实例属性
field_str = ""
for i in field:
field_str += i + "," # 将所有的字段与字段类型以 “ , ” 拼接
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: # 判断该表是否已存在
return # 该表已存在！直接返回
field_str = field_str.rstrip(",") # 删除字符串最后的“,”
sql = f"""
create table {self.table_name}(
id int primary key auto_increment,
{field_str}
);
"""
self.mysql.execute(sql)
except Exception as e:
print("创建表失败！")
print("错误信息：", e)
else:
print("创建表成功！")
现在我们就实现了创建表！
效果相当于定义模型类，并且迁移成功！
并且实现了每张表创建时自动添加id字段！
id字段类型为主键，自增。
打印一下最终执行的语句sql：
四个字段，完美！
其实就把传过来的字段列表，循环拼接每个字段就可以了！
同时获取表名、字段名，方便后面进行增删改查。
下面我就直接发代码了！
模型层实现
模型层所有代码
import pymysql
# mysql操作
class Mysql():
def __init__(self,host=None,port=None,user=None,password=None,database=None,charset=None):
try:
self.table_name = None # 表名
self.field = ["id",] # 表字段
self.pymysql_con = pymysql.connect(host=host,port=port,user=user,password=password,database=database,charset=charset)
self.mysql = self.pymysql_con.cursor() # 创建游标对象
except Exception as e:
print("连接数据库失败！")
print("错误信息：",e)
else:
print("连接数据库成功！")
# 获取数据库内所有的表
def show_tables(self):
try:
sql = f"""show tables;"""
self.mysql.execute(sql)
except Exception as e:
print("查询表失败！")
print("错误信息：", e)
else:
data = self.mysql.fetchall()
table_list = []
for i in data: # 元组转换为列表
table_list.append(i[0])
return table_list # 返回当前数据库内所有的表
# 创建表 接收一个字符串类型的表名，和一个列表、元组类型的表字段
def create_table(self, table_name=None, field=None):
try:
self.table_name = table_name # 将表名赋值给实例属性
field_str = ""
for i in field:
field_str += i + "," # 将所有的字段与字段类型以 “ , ” 拼接
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: # 判断该表是否已存在
return # 该表已存在！直接返回
field_str = field_str.rstrip(",") # 删除字符串最后的“,”
sql = f"""
create table {self.table_name}(
id int primary key auto_increment,
{field_str}
);
"""
print(sql)
self.mysql.execute(sql)
except Exception as e:
print("创建表失败！")
print("错误信息：", e)
else:
print("创建表成功！")
# 查询数据
def all(self):
try:
sql = f"""select * from {self.table_name};""" # 根据表名直接查询
self.mysql.execute(sql)
except Exception as e:
print("查询失败！")
print("错误信息：", e)
else:
data = self.mysql.fetchall()
data_list = []
for i in data: # 进行数据转换
data_dict = {}
for k,j in enumerate(self.field):# 每行数据 组成字典类型
data_dict[j] = i[k]
data_list.append(data_dict)
return data_list # 最终返回查询集
# 添加数据 接收所有的命名参数 即根据字段名传入对应数据
def create(self,**kwargs):
try:
create_data = ""
for i in self.field:
if i == "id": # id 字段跳过
continue
try:
if not kwargs[i]: # 传入的改字段值为空
print("错误信息：",i,"字段为空")
return False
except Exception: # 异常时缺少字段
print("字段不全！")
return False
if type(kwargs[i]) == str: # 判断类型 为字符串 需加引号
create_data += "'" + str(kwargs[i]) + "',"
else:
create_data += str(kwargs[i]) + ","
field_str = create_data.rstrip(",") # 删除字符最后的“,”
# id 字段为null ，默认自增
sql = f"""
insert into {self.table_name} values
(null ,{field_str});
"""
self.mysql.execute(sql)
except Exception as e:
self.pymysql_con.rollback()
print("添加失败！")
print("错误信息：", e)
return False
else:
self.pymysql_con.commit()
print("添加成功！")
return True
# 修改数据 接收所有的命名参数 即根据字段名传入对应数据
def update(self, **kwargs):
try:
update_str = ""
for i in self.field:
try:
if not kwargs[i]: # 传入字段值参数为空 警告 字段值为空则不做修改
print("警告：", i, "传入该字段值为空，则该字段不做修改！")
continue
except Exception: # 缺少字段，该字段也不做修改
print("提示：", i, "该字段没有传入参数，则该字段不做修改！")
continue
if type(kwargs[i]) == str: # 判断类型
update_str += i + "='" + str(kwargs[i]) + "',"
else:
update_str += i + "=" + str(kwargs[i]) + ","
field_str = update_str.rstrip(",") # 删除字符最后的“,”
# 根据传入id 进行修改数据 若没有传入id 则报错 修改失败
sql = f"""
update {self.table_name} set {field_str} where id = {kwargs["id"]};
"""
self.mysql.execute(sql)
except Exception as e:
self.pymysql_con.rollback()
print("修改失败！")
print("错误信息：", e)
return False
else:
self.pymysql_con.commit()
print("修改成功！")
return True
# 删除数据的方法 接收所有的命名参数 即根据字段名传入对应数据
def delete(self, **kwargs):
try:
delete_str = ""
for i in self.field:
try:
if not kwargs[i]: # 字段值为空 返回删除失败！
print("错误信息：",i,"该字段值为空！")
return False
except Exception:# 该字段没有传入跳过
continue
if type(kwargs[i]) == str: # 判断类型
delete_str += i + "='" + str(kwargs[i]) + "' and "
else:
delete_str += i + "=" + str(kwargs[i]) + " and "
field_str = delete_str.rstrip(" and ") # 删除字符最后的“,”
# 先查询满足条件的数据个数
sql = f"""
select * from {self.table_name} where {field_str};
"""
self.mysql.execute(sql)
data = self.mysql.fetchall()
if not data: # 没有查询到则返回删除失败
print("没有可删除的数据！")
print("删除失败！")
return False
# 删除数据
sql = f"""
delete from {self.table_name} where {field_str};
"""
self.mysql.execute(sql)
except Exception as e:
self.pymysql_con.rollback()
print ("删除失败！")
print("错误信息：", e)
return False
else:
self.pymysql_con.commit()
print("删除成功！")
return True
# 过滤查询数据
def filter(self,**kwargs):
try:
filter_str = ""
for i in self.field:
try:
if not kwargs[i]: # 传入字段值参数为空 警告
print("警告：", i, "传入该字段值为空！")
continue
except Exception:
continue
if type(kwargs[i]) == str:
filter_str += i + "='" + str(kwargs[i]) + "' and "
else:
filter_str += i + "=" + str(kwargs[i]) + " and "
field_str = filter_str.rstrip(" and ") # 删除字符最后的“,”
sql = f"""
select * from {self.table_name} where {field_str};
"""
self.mysql.execute(sql)
except Exception as e:
print("查询失败！")
print("错误信息：", e)
else:
data = self.mysql.fetchall()
data_list = []
for i in data:
data_dict = {}
for k,j in enumerate(self.field):
data_dict[j] = i[k]
data_list.append(data_dict)
# 返回查询集
return data_list
# 过滤获取数据 返回第一个数据
def get(self,**kwargs):
try:
get_str = ""
for i in self.field:
try:
if not kwargs[i]:
print("错误信息：",i,"字段为空")
return False
except Exception:
continue
if type(kwargs[i]) == str:
get_str += i + "='" + str(kwargs[i]) + "' and "
else:
get_str += i + "=" + str(kwargs[i]) + " and "
field_str = get_str.rstrip(" and ") # 删除字符最后的“,”
sql = f"""select * from {self.table_name}
where {field_str};
"""
print(sql)
self.mysql.execute(sql)
except Exception as e:
print("查询失败！")
print("错误信息：", e)
else:
data = self.mysql.fetchall()
data_list = []
for i in data:
data_dict = {}
for k,j in enumerate(self.field):
data_dict[j] = i[k]
data_list.append(data_dict)
if data_list: # 只返回查询到的第一行数据
return data_list[0] # dict 字典类型
return {}
视图使用
使用说明
from flask_app import creat_flask_app
from flask_cors import CORS
from flask import request
from settings import DefaultConfig
app = creat_flask_app(DefaultConfig)
import json
class_data = app.config.get("CLASS_LIST") # 从配置文件中读取班级信息
student_data = app.config.get("STUDENT_LIST") # 从配置文件中读取学生信息
user = app.config.get("MYSQL_USER")
password = app.config.get("MYSQL_PASSWORD")
host = app.config.get("MYSQL_HOST")
port = app.config.get("MYSQL_PORT")
name = app.config.get("MYSQL_DATABASE")
charset = app.config.get("MYSQL_CHARSET")
# 导入封装的mysql
from Mysql import Mysql
# 实例mysql对象
student = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# 实例一个对象，相当于创建一个模型类，可实例多个模型类
# teacher = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# goods = Mysql(host=host,port=port,user=user,password=password,database=name,charset=charset)
# 表字段
student_table_fields = [
"name varchar (20)", # 一个字符串为一个字段
"age int",
"gender enum('男','女')",
"phone varchar(11)",
]
# 调用创建表方法 第一个参数为表名， 第二个参数为表字段
student.create_table("student_1",student_table_fields)
# 添加数据
@app.route("/add/",methods=["POST"])
def add_data():
names = request.form.get("name")
ages = request.form.get("age")
genders = request.form.get("gender")
phones = request.form.get("phone")
# 添加数据 # 字段 需跟创建表字段时相同
flag = student.create(name=names,age=ages,gender=genders,phone=phones)
# 返回 true 为成功
# 返回 false 失败
if flag:
return "添加成功！"
else:
return "添加失败！"
# 获取全部数据
@app.route("/list/",methods=["GET"])
def student_list():
data = student.all() # 获取所有信息
ser = json.dumps(data)
return ser
# 修改数据
@app.route("/update/",methods=["PUT"])
def student_update():
pk = request.form.get("id")
names = request.form.get("name")
ages = request.form.get("age")
genders = request.form.get("gender")
phones = request.form.get("phone")
# 修改数据 # 根据id修改后面的数据
flag = student.update(id=pk,name=names, age=ages, gender=genders, phone=phones)
# 表字段值允许为空， 为None该字段不做修改
# flag = student.update(id=pk,name=names, gender=genders, phone=phones)
# 如 names 值为 None 也不做修改，该字段不做修改
if flag:# true 为成功
return "修改成功！"
else:
return "修改失败！"
# 删除数据
@app.route("/delete/",methods=["DELETE"])
def student_delete():
pk = request.form.get("id")
names = request.form.get("name")
# 根据表字段删除
flag = student.delete(id=pk,name=names)
# 例：删除 id = 1 并且 name = "张三" 的数据
# 条件成立：删除数据，返回 true
# 条件不成立：没有删除数据，返回 false
if flag:# true 为成功
return "删除成功！"
else:
return "删除失败！"
# 过滤
@app.route("/filter/",methods=["GET"])
def filter_data():
names = request.form.get("name")
ages = request.form.get("age")
# 根据表字段进行过滤查询
data = student.filter(age=ages)
# 条件成立：返回查询集 [...]
# 条件不成立：返回 []
ser = json.dumps(data)
return ser
# 获取单个数据
@app.route("/get/",methods=["GET"])
def get_data():
names = request.form.get("name")
# 根据表字段过滤查询
data = student.get(name=names)
# 条件成立：查询成功，返回第一个数据
# 条件不成立：返回空 {}
ser = json.dumps(data)
return ser
if __name__ == '__main__':
app.run(port=8000)
以上就是模型层的实现及使用啦！
总结如果觉得还不错，那就点个赞吧！
欢迎大家使用！如果发现有什么bug欢迎在评论区留言！
