在做接口自动化时往往需要初始化数据库操作，处理脏数据。数据初始化主要包括：
- 数据库连接
- 数据清除
- 数据插入
- 关闭数据库
实例
首先创建：mysql_action.py文件
mysql_action.py
from pymysql import connect
class DB():
def __init__(self):
'''连接数据库'''
self.conn = connect(host="127.0.0.1",user="root",password="123456",db="django_restful",port=3306)
def clear(self, table_name):
'''清除表中数据'''
# sql语句
clear_sql = 'truncate ' + table_name + ';' # 注意两个字符串拼接中间有个空格
with self.conn.cursor() as cur:
# 清除外键约束
cur.execute('set foreign_key_checks=0;')
# 加载sql
cur.execute(clear_sql)
# 执行
self.conn.commit()
def insert(self, table_name, table_data):
'''插入数据'''
# 遍历数据，把value值加上单引号
for key in table_data:
table_data[key] = "'" + str(table_data[key]) + "'"
# 把key值和value值通过逗号连接
key = ','.join(table_data.keys())
value = ','.join(table_data.values())
# 插入sql
insert_sql = 'insert into ' + table_name + '(' + key + ')' + 'values' + '(' + value + ')'
# 加载sql
with self.conn.cursor() as cur:
cur.execute(insert_sql)
# 执行
self.conn.commit()
def close(self):
'''关闭连接'''
self.conn.close()
def init_data(self, datas):
'''初始化数据'''
for table, data in datas.items():
# 清除表数据
self.clear(table)
# 插入新数据
self.insert(table, data)
self.close()
if __name__ == '__main__':
'''调试'''
db = DB()
# 清除表数据
# db.clear('api_user')
# 插入数据
# data = {'id': 1, 'username': 'aa', 'email': 'aa@163.com', 'groups': '1'}
# db.insert('api_user',**data)
# 关闭连接
# db.close()
# 初始化数据{表名: {表数据}}
table_data = {'api_user': {'id': 1, 'username': 'aa', 'email': 'aa@163.com', 'groups': '1'}}
db.init_data(table_data)
其次封装初始化数据。
如上代码可以看出，把数据和代码写在一起不方便维护。我们将初始化数据使用YamI来封装，可以将数据与脚本分离，方便测试数据的维护。
创建一个datas.yaml文件，数据内容如下：
# 表名1
api_user:
# 表中数据
- id: 1
username: aa
email: aa@163.com
groups: 1
- id: 2
username: bb
email: bb@163.com
groups: 2
# 表2
api_group:
- id: 1
name: A
- id: 2
name: B
重新封装初始化数据函数
安装yaml模块：pip install pyyaml
修改后的代码如下
from pymysql import connect
import yaml
class DB():
def __init__(self):
'''连接数据库'''
print('connect db...')
self.conn = connect(host="127.0.0.1",user="root",password="123456",db="django_restful",port=3306)
def clear(self, table_name):
'''清除表中数据'''
# sql语句
clear_sql = 'truncate ' + table_name + ';' # 注意两个字符串拼接中间有个空格
with self.conn.cursor() as cur:
# 清除外键约束
cur.execute('set foreign_key_checks=0;')
# 加载sql
cur.execute(clear_sql)
# 执行
self.conn.commit()
def insert(self, table_name, table_data):
'''插入数据'''
# 遍历数据，把value值加上单引号
for key in table_data:
table_data[key] = "'" + str(table_data[key]) + "'"
# 把key值和value值通过逗号连接
key = ','.join(table_data.keys())
value = ','.join(table_data.values())
# 插入sql
insert_sql = 'insert into ' + table_name + '(' + key + ')' + 'values' + '(' + value + ')'
# 加载sql
with self.conn.cursor() as cur:
cur.execute(insert_sql)
# 执行
self.conn.commit()
def close(self):
'''关闭连接'''
self.conn.close()
def init_data(self, datas):
'''初始化数据'''
print(datas)
# 清除表数据
for table_name, table_data in datas.items():
self.clear(table_name)
# 插入新数据
for data in table_data:
self.insert(table_name, data)
self.close()
if __name__ == '__main__':
# 初始化数据
with open('datas.yaml', 'r') as f:
datas = yaml.load(f)
db.init_data(datas)
执行结果
{'api_group': [{'name': 'A', 'id': 1}, {'name': 'B', 'id': 2}], 'api_user': [{'username': 'aa', 'email': 'aa@163.com', 'groups': 1, 'id': 1}, {'username': 'bb', 'email': 'bb@163.com', 'groups': 2, 'id': 2}]}
api_user表
api_group表
