什么是 「PyMySQL」?PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库,Python2 中则使用 mysqldb。
1 使用的目的
- 测试前操作数据库准备测试数据。
- 测试后查询数据库数据入库情况。
- 测试结束以后清除测试数据,恢复之前的状态。
2 PyMySQL 安装
- $ pip3 install PyMySQL
❝以下来自菜鸟教程
- 实例:
❞
实例(Python 3.0+)
#!/usr/bin/python3
import pymysql
# 打开数据库连接
db = pymysql.connect(host='localhost',
user='testuser',
password='test123',
database='TESTDB')
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
# 使用 execute() 方法执行 SQL 查询
cursor.execute("SELECT VERSION()")
# 使用 fetchone() 方法获取单条数据.
data = cursor.fetchone()
print ("Database version : %s " % data)
# 关闭数据库连接
db.close()
- 可以看到查询需要用到的信息,主要是数据库ip,用户名,密码,数据库名,还有没有列出来的端口号
- 这些我们可以配置到yaml文件里面维护
3 使用yaml文件配置数据库连接信息
我们先建一个表,字段分别是id
,busi_order
订单,state
状态,crate_time
创建时间,describe
描述。
- 创建一个工具类
#-*- codeing = utf-8 -*-
#@Time: 2022/10/30 19:01
#@Author: 怪盗LLYL
#@File: db_unit.py
#@Software: PyCharm
import pymysql
from conf.config import FilePath
from unit.conf_yaml import YamlReader
class DateBase(object):
#初始化
def __init__(self,env,yamlf_name=FilePath.DB_CONF_FILE):
self._connect_conf = YamlReader(yamlf_name).data()[env]
def basedb(slef,key:str):
'''根据key选择数据库表
:param key:订单号最后一位
:return 数据库表连接:type:pymysql.connections.Connection
'''
key=int(key)
database_count = len(slef._connect_conf['db_order']['database'])
selected_database = (key)%database_count
return pymysql.connect(host=slef._connect_conf['db_order']['host'],
port=slef._connect_conf['db_order']['port'],
user=slef._connect_conf['db_order']['user'],
password=slef._connect_conf['db_order']['password'],
database=slef._connect_conf['db_order']['database'][selected_database])
- 这里假设我们有两个库,一库和二库,根据尾数分别入了不同的库里,我们可以初选择不同数据库连接信息,具体情况请根据读者公司实际逻辑编写。
- YamlReader类请看上一篇文章。
4 添加一个查询方法:
「注意datetime字段需要特殊处理下, cursor.description 是用来获取列名的。」
def select(slef,db,sql):
# 打开数据库连接
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
# 使用 execute() 方法执行 SQL 查询
cursor.execute(sql)
# 使用 fetchone() 方法获取单条数据.
# 使用 fetchall() 方法获取所有数据.
result = cursor.fetchall()
data_list = []
for z in result:
i = 0
data_dict = {}
for field in cursor.description:
if isinstance(z[i], datetime.datetime) or isinstance(z[i], datetime.date):
data_dict[field[0]] = str(z[i])
else:
data_dict[field[0]] = z[i]
i = i + 1
data_list.append(data_dict)
# 关闭数据库连接
db.close()
return data_list
def get_date_by_busi_order(slef,busi_order:str,table:str):
'''根据busi_order查询订单信息
:param busi_order:订单号
:param table:表名
:return df List[dic]:返回订单信息,字典内key:字段 value:值
'''
key = int(busi_order[-1])
db=slef.basedb(key)
sql=f"select * from {table} where busi_order='{busi_order}'"
df=slef.select(db,sql)
return df
- 先手动插入数据
if __name__ == '__main__':
datebase=DateBase("dev")
busi_order = "test123456789"
table = "table_order"
df = datebase.get_date_by_busi_order(busi_order=busi_order,table=table)
print(df)
busi_order = "test12345678"
table = "table_order"
df = datebase.get_date_by_busi_order(busi_order=busi_order,table=table)
print(df)
- 可以看到两条不同库的数据都查出来了
5 插入操作:
- 创建插入方法,其中订单数据我临时使用到一个字典数据,读者可以按照实际情况修改。
def insert(self,db,sql):
'''根据db和sql执行插入操作
:param db:数据库连接
:param sql:sql语句 插入语句
:return code int:操作结果 200:成功 500:失败
'''
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
# 使用 execute() 方法执行 SQL 查询
code = 0
try:
# 执行sql语句
cursor.execute(sql)
# 提交到数据库执行
db.commit()
code = 200
except:
# 如果发生错误则回滚
db.rollback()
code = 500
# 关闭数据库连接
db.close()
return code
def insert_order(slef,table_order:dict,table:str):
'''根据table_order订单信息插入数据库
:param table_order dir:订单信息
:param table:表名
:return code int:操作结果 200:成功 500:失败
'''
print(table_order)
key = int(table_order["busi_order"][-1])
db=slef.basedb(key)
sql=f"INSERT INTO {table} "+str(tuple(table_order.keys())).replace("'","`")+" VALUES "+str(tuple(table_order.values())) #注意mysql语句中字段不是单引号而是反引号
code=slef.insert(db,sql)
return code
if __name__ == '__main__':
datebase=DateBase("dev")
table_order = {"busi_order":"test654321721","state":"200","create_time":"2022-10-30 20:06:25","describe":"测试插入数据库"}
table = "table_order"
code = datebase.insert_order(table_order=table_order,table=table)
print(code)
- 可以看到拼接了插入语句,插入成功返回200。
- 查看数据库,插入成功了
- 再次执行因为订单号重复的原因插入失败了,并且没有入数据库
6 更新和删除操作:
- 更新,删除和插入没有什么不同,所以可以把insrt改成update其他方法都调用这个。
def update (self,db,sql):
'''根据db和sql执行插入,更新,删除操作
:param db:数据库连接
:param sql:sql语句 执行语句
:return code int:操作结果 200:成功 500:失败
'''
# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
# 使用 execute() 方法执行 SQL 查询
code = 0
try:
# 执行sql语句
cursor.execute(sql)
# 提交到数据库执行
db.commit()
code = 200
except Exception as e:
# 如果发生错误则回滚
print("sql出错:",sql)
print(e.__str__())
db.rollback()
code = 500
# 关闭数据库连接
db.close()
return code
def insert_order(slef,table_order:dict,table:str):
'''根据table_order订单信息插入数据库
:param table_order dir:订单信息
:param table:表名
:return code int:操作结果 200:成功 500:失败
'''
key = int(table_order["busi_order"][-1])
db=slef.basedb(key)
sql=f"INSERT INTO {table} "+str(tuple(table_order.keys())).replace("'","`")+" VALUES "+str(tuple(table_order.values())) #注意mysql语句中字段不是单引号而是反引号
code=slef.update(db,sql)
return code
def update_order(slef,table_order:dict,table:str):
'''根据table_order订单信息修改记录
:param table_order dir:订单信息
:param table:表名
:return code int:操作结果 200:成功 500:失败
'''
key = int(table_order["busi_order"][-1])
db=slef.basedb(key)
sql = f"UPDATE {table} SET "
for i in table_order:
sql = sql + f"`{i}`= '{str(table_order[i])}',"#注意mysql语句中字段不是单引号而是反引号
sql = sql[:-1]
sql = sql + f" WHERE busi_order = '{table_order['busi_order']}';"
code=slef.update(db,sql)
return code
def delete_order(slef,table_order:dict,table:str):
'''根据table_order订单信息删除记录
:param table_order dir:订单信息
:param table:表名
:return code int:操作结果 200:成功 500:失败
'''
key = int(table_order["busi_order"][-1])
db=slef.basedb(key)
sql = f"DELETE FROM {table} WHERE busi_order = '{table_order['busi_order']}';"
code=slef.update(db,sql)
return code
if __name__ == '__main__':
datebase=DateBase("dev")
table_order = {"busi_order":"test654321721","state":"200","create_time":"2022-10-30 22:13:25","describe":"更新操作"}
table = "table_order"
code = datebase.update_order(table_order=table_order,table=table)
print(code)
- 更新成功,插入成功,删除成功
7 结语
- 这样pymysql入门操作就介绍完了。大家可以根据自己实际情况在自己的项目里使用,或者使用orm。
未经允许不得转载:便宜VPS » 测试中简单的数据库操作-pymysql入门