一、Mysql数据库封装

在common中新建一个mysql_db_util.py的文件

import pymysqlclass MysqlDBUtil:#字典输出def __init__(self, host, user, password, port=3306):self.connect = pymysql.Connect(host=host,port=port,user=user,password=password,charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)def exe_sql(self, sql):try:# 获取游标对象cursor = self.connect.cursor()# 调用游标对象的execute方法，执行sqlcursor.execute(sql)#  如果是查询if sql.split()[0].lower() == "select":# 返回所有数据return cursor.fetchall()#  否则：else:# 提交事务self.connect.commit()# 返回受影响的行数return cursor.rowcountexcept Exception as e:# 事务回滚self.connect.rollback()# 打印异常信息print(e)finally:# 关闭游标cursor.close()# 关闭连接self.connect.close()if __name__ == '__main__':db_util = MysqlDBUtil(host='localhost', user='root', password='123456')result = db_util.exe_sql('select * from community.tb_user')print(result)

输出结果如下：

发贴的测试用例添加数据库的断言校验以及连接数据库通过配置文件获取数据库的账号、密码等信息；

import pytestfrom api.portal.portalArticle.create_article_api import CreateArticleApi
from common.file_load import read_excel, load_yaml_file
from common.mysql_db_util import MysqlDBUtilclass TestCreateArticleApi:test_data = read_excel('\data\community_testdata.xlsx', '发帖数据')# test_data = [#     ['标题为空', '', '超级管理员', '内容', '1', 200, '操作成功'],#     ['正常发帖', '1111', '超级管理员', '内容', '1', 200, '操作成功'],#     ['发帖用户为空', '1111', '', '内容', '1', 200, '操作成功'],# ]@pytest.mark.parametrize('casename,title, author, content, createUser,expect_code,expect_message', test_data)def test_buy_now_params(self, casename, title, author, content, createUser, expect_code, expect_message):createArticleApi = CreateArticleApi(title, author, content, createUser)resp = createArticleApi.sendRequest()print(resp.text)status_code = resp.status_codeassert status_code == expect_codemessage = resp.json()['msg']assert message == expect_message# 通过yml获取DB的信息db_info = load_yaml_file('/config/db.yml')['community']host = db_info['host']userName = db_info['userName']password = db_info['password']port = db_info['port']mysqlDBUtil = MysqlDBUtil(host=host, user=userName, password=password,port=port)sqlResult = mysqlDBUtil.exe_sql(f"select count(1) as nub from community.tb_article where title = '{title}'")[0]['nub']assert sqlResult != 0

二、redis数据库封装

新建一个redis_util.py

import javaobj
import redisclass RedisUtil:def __init__(self,host,pwd,port=6379,decode_responses=False,db=0):# 先创建一个连接池self.pool = redis.ConnectionPool(host=host, password=pwd,port=port, decode_responses=decode_responses, db=db)self.r = redis.Redis(connection_pool=self.pool)  # 从连接池里拿到一个连接对象# 通过连接对象，对redis进行操作def get(self,key):# 判断该key对应的数据类型是什么，然后决定要用哪个数据获取方法type = self.r.type(key).decode('utf-8')if type == 'string':return self.r.get(key)elif type == 'hash':return self.r.hgetall(key)elif type == 'list':return self.r.lrange(key,0,-1)elif type == 'set':return self.r.smembers(key)elif type == 'zset':return self.r.zrange(key,0,-1)else:raise Exception(f'{key}对应的数据类型不支持')if __name__ == '__main__':redis_util = RedisUtil(host='localhost',pwd='123456')res = redis_util.get('{XXX_XXX_XXX}_59') #59是用户id# res得到的数据是项目向缓存中存储的一个java对象的序列化数据，我们无法直接从中得到想要的数据# 所以我们需要先将该数据转换成python对象，然后再处理print(res)res_obejct = javaobj.loads(res)print(res_obejct)print(type(res_obejct))print(res_obejct[0])# 获取该对象下都有哪些属性信息,看到的业务属性需要和开发沟通是什么意思print(dir(res_obejct[0]))# 获取立即购买时存储到redis的num数量print(res_obejct[0].__getattribute__('num'))print(res_obejct[0].__getattribute__('skuId'))