python+pymysql调用数据库准备数据
2017-05-22 本文已影响410人
赫本iii
测试人员是不是经常因为一些测试需要的数据进行烦恼。
例如：开发人员需要每个状态的用户，而你需要不厌其烦的每个账号去申请，创造每个状态的用户。其实像这种重复劳动损伤脑细胞的活为何不交给机器去完成呢。
我们进入正题
1. 研读pymysql的API文档（二次封装）
- 连接数据库（创建游标）
- 准备sql（需要变量的参数化)
- 执行sql语句的方法
- 获取自增ID的方法
- 查询语句方法
2.业务代码（各个状态的账号）
3.框架实现
- 调用二次封装的框架
- common文件对应的sql语句
- 调用业务代码生成账号
一、写一个封装的调用mysql的模块
代码实现如下：
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 数据准备调用mysql
import pymysql.cursors
import logging
import uuid
import time
import common as c
DB = None
def connect(host, user, password, db, port,charset='utf8'):
"""连接MySql"""
global DB
if DB == None:
DB = pymysql.connect(host=host,
user=user,
password=password,
db=db,
port=port,
charset=charset,
cursorclass=pymysql.cursors.DictCursor)
return DB
def execute(sql):
# type: (object) -> object
"""执行sql
:return row number affected
:except sql error
"""
global DB
try:
with DB.cursor() as cursor:
res = cursor.execute(sql)
DB.commit() # 链接不会默认提交需要您手动提交
return res
except Exception, e:
DB.rollback() # 若错误则回滚
logging.info("sql is empty or error %s", e)
def getNewId(sql):
"""执行sql获取自增ID"""
global DB
try:
with DB.cursor() as cursor:
cursor.execute(sql)
DB.commit()
res = cursor.lastrowid
# 获取最新的语句的自增ID
return res
except Exception, e:
DB.rollback() # 若错误则回滚
logging.info("sql is empty or error %s", e)
def select(sql):
"""查询语句获取查询结果"""
global DB
try:
with DB.cursor() as cursor:
cursor.execute(sql)
result = cursor.fetchone()
print result
except Exception, e:
logging.info("sql is empty or error %s", e)
def setUid():
"""生成一个唯一的UID"""
uid = str(uuid.uuid1())
id = ''.join(uid.split('-'))
return id
def setPhone():
"""生成一个唯一手机号码"""
number = time.time()
phone = int(number * 10)
return phone
def setEmail():
"""生成一个唯一的邮箱号码"""
number = str(int(time.time()))
email = number[-5:] + c.EMAIL
return email
def close():
"""关闭MySQL连接"""
global DB
DB.close()
上面用到了一些python内置库生成手机号和邮箱，还有用户所需的uid。
二、引用log模块获取日志
准备工作：
需要一个日志的捕获，包括框架和源码抛出的expection。
代码如下：
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 基础包：日志服务
import logging
import time
def getLogger():
global tezLogPath
try:
tezLogPath
except NameError:
tezLogPath = "/data/log/apiTest/"
FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
# file = tezLogPath + time.strftime("%Y-%m-%d", time.localtime()) + ".log"
# logging.basicConfig(filename=file, level=logging.INFO, format=FORMAT)
# 开发阶段为了方便调试，可不输出到文件
logging.basicConfig(level=logging.INFO, format=FORMAT)
return logging
三、构建业务逻辑代码
准备工作：
业务逻辑相关生成各种状态的代码
代码如下：
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 数据准备所需代码
import dataPreparationSql as sql
import common as c
import log
logging = log.getLogger()
host = c.HOST
user = c.USER
password = c.PASSWORD
db = c.DB
port = c.PORT
email = sql.setEmail()
phone = sql.setPhone()
uid = sql.setUid()
connect = sql.connect(host=host, user=user, password=password, db=db, port=port)
def addProUser(proName, userEmail, nickName, companyId=None):
"""
新增企业用户
:Args:
proName - 企业名称
userEmail - 企业邮箱
nickName - 用户昵称
companyID - 默认为空重新生成，需要添加指定企业ID需手动填写参数
"""
global connect
try:
if companyId == None:
company = sql.getNewId(c.SQL_COMPANY % proName) # 执行sql新增一个企业 tbl_pro_company
sql.execute(c.SQL_USER % (userEmail, phone, uid)) # 执行sql增加一个用户 tbl_user
res = sql.execute(c.SQL_PROUSER % (uid, company, nickName, userEmail, phone)) # 执行sql添加一个企业用户 tbl_pro_user
# select = sql.select((c.SQL_SELECT) % companyId)
logging.info("effect row %s", res)
# logging.info("results %s", select)
else:
sql.execute(c.SQL_USER % (userEmail, phone, uid))
res = sql.execute(c.SQL_PROUSER % (uid, companyId, nickName, userEmail, phone))
logging.info("effect row %s", res)
except Exception, e:
logging.info("add Pro User error %s", e)
def addCommonUser(userEmail, nickName):
"""
新增普通客户
:Args:
userEmail - 用户邮箱
nickName - 用户昵称
"""
global connect
try:
accountId = sql.getNewId(c.SQL_ACCOUNTID % uid) # 添加一个账户ID tbl_account
sql.execute(c.SQL_COMMON_USER % (userEmail, phone, uid, accountId)) # 添加一个用户到tbl_user
sql.execute(c.SQL_CUSTOMER % (uid, nickName, userEmail, phone)) # 添加一个用户到tbl_customer
res = sql.execute(c.SQL_WALLET % (uid, accountId)) # 添加一个钱包表
logging.info("effect row : %s", res)
except Exception, e:
logging.info("add Common User error: %s", e)
def addDesigner(userEmail, nickName):
"""
新增普通设计师
:Args:
userEmail - 用户邮箱
nickName - 用户昵称
"""
global connect
try:
accountId = sql.getNewId(c.SQL_ACCOUNTID % uid) # 添加一个账户ID tbl_account
sql.execute(c.SQL_DESIGNER_USER % (userEmail, phone, uid, accountId)) # 添加一个用户到tbl_user
sql.execute(c.SQL_DESIGNER % (uid, nickName, userEmail, phone)) # 添加一个用户到tbl_designer
sql.execute(c.SQL_WALLET % (uid, accountId)) # 添加一个钱包表
remarkId = sql.getNewId(c.SQL_REMARK % uid) # 添加一个评价ID 不然会影响推送
res = sql.execute(c.SQL_ABILITY % (remarkId, uid)) # 添加一个设计师能力评分 不然影响推送
logging.info("effect row : %s", res)
except Exception, e:
logging.info("add designer error: %s", e)
四、关于common模块
准备工作：
所有的参数和常量我们会整理到这个文件中，因为设计业务和服务密码、数据库密码这里展示一部分。
代码如下：
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# sql常量
#测试环境数据库配置
HOST = '127.0.0.1'
USER = 'root'
PASSWORD = '123456'
PORT = 3306
DB = 'MYSQL_DB'
EMAIL = '@sina.com'
sql = 'SELECT * FROM tbl_user'
SQL_COMPANY = """
INSERT INTO tbl_pro_company (key) VALUE (values)
"""
SQL_USER = """
INSERT INTO tbl_user (key) VALUE (values)
"""
SQL_PROUSER = """
INSERT INTO tbl_pro_user (key) VALUE (values)
"""
五、写一个run文件安全性剥离
需要注意为什么会用time.sleep()，因为我们用了python内置库生成uid根据时间戳生成md5格式的。如果直接运行这3个账号的uid就一样了，违背了uid唯一性了。
代码如下：
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 执行脚本生成数据
import tezPreData as Data
import time
Data.addDesigner('账号','名称') # 添加一个设计师
time.sleep(2)
Data.addProuser('企业名称','账号','名称') # 添加一个企业用户
time.sleep(2)
Data.addCommonUser('账号','名称') # 添加一个普通用户
time.sleep(2)
Data.sql.close()
那么一个简易的数据准备脚本完成，可以按照自己业务所需的情况，调用函数。请告别手工，解放双手才是一个IT工程师最应该去关注的。
