Python3 从零单排14_flask模块&mysql操作封装

　　1.flask模块介绍

　　在测试的过程中，往往会遇到功能受接口影响而导致测试无法正常进行，往往要等接口完成之后才能做功能测试，但这往往很托节奏。这个时候就可以考虑，将接口和功能剥离，分开测试，同时进行，保证项目进度。

　　问题是怎么分开呢？要研发先帮忙开发接口、调试好，然后测试？显然太low，python这个时候就可以来装一波X，我们自己动手按照需求开发一套接口来供功能测试，其第三方模块flask提供接口服务，早就有大神开发好接口服务的第三方模块，我们直接用就可以了，这就是python的强大之处了，flask模块使用如下：

import flask  #导入模块
server=flask.Flask(__name__)  #创建接口服务，格式：flask.Flask(__name__),类似于redis.Redis,其中(__name__)是指当前的python文件，也就是说接口是基于此文档在运行
@server.route('/api',methods=['get','post'])#创建好接口服务后，需要指定路径，路径包含后缀以及请求方法，必须要是@装饰，下面的函数才能正常运行
def reg():#接口主体部分，运行内容
    return ('hello world!')
server.run(port=9999)#服务创建好，并指定好路径，接口内容编写完后，需要启动服务，启动的时候需要指定端口

　　上面就是最简洁的接口服务了，运行后就可以访问了，在浏览器输入：http://127.0.0.1:9999/api，浏览器会输出：“hello world!”。但如果想要获取到用户的请求数据、以及局域网内别的测试人员要访问你的接口咋办，代码如下：

import flask  #导入模块
from flask import request#导入request方法，用来接收传参
server=flask.Flask(__name__)  #创建接口服务，格式：flask.Flask(__name__),类似于redis.Redis,其中(__name__)是指当前的python文件，也就是说接口是基于此文档在运行
@server.route('/login',methods=['get','post'])#创建好接口服务后，需要指定路径，路径包含后缀以及请求方法，必须要是@装饰，下面的函数才能正常运行
def reg():#接口主体部分，运行内容
    name=request.values.get("username")#request.values.get("username")获取指定传参的value，用来判断、处理
    passwd=request.values.get("password")#如果传参是json类型，方法是request.json.get("username")
    if name and passwd:
        if name.strip() and passwd.strip():
            return '{"code":200,"msg":"注册成功！"}'
    return '{"code":300,"msg":"参数不能为空！"}'
server.run(port=9999,host='0.0.0.0',debug=True)#指定host为“0.0.0.0”后，局域网内其他IP就都可以访问了

　　上述代码基本可以满足接口需求了，然后你可以在接口主体内容内，对传参进行分析、处理，比如得到传参后，与数据库里的数据进行比对、分析，给出正确响应。譬如用户输入用户名和密码，取到参数后，查询数据库中是否存在用户名，且用户名密码是否正确，用来判断是否允许登录。既然如此那么我们要初始化一些数据了，就是在运行接口之前要判断数据库、表以及数据是否存在，不存在则需要初始化创建上述一系列数据，创建数据库的函数如下：

　　2.mysql操作函数封装

def oprate_sql(sql, HOST, USER, PASSWD, charset='utf8', port=3306):#定义操作sql语句的函数，如果是select或show则返回查询内容
    import pymysql
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWD, charset=charset)
    cur = conn.cursor()
    cur.execute(sql)
    if sql.strip().startswith('select') or sql.strip().startswith('show'):
        res = cur.fetchall()
    else:
        conn.commit()
        res = 'ok'
    cur.close()
    conn.close()
    return res

def check_in(args,res):#判断a,是否在res的元素里，这里的res必须是sql执行返回的数据,用于注册，登录
    lis=[]
    for i in res:
        lis.append(i[0])
    if args in lis:
        return True
#调用：
sql='show databases;'
res=oprate_sql(sql, HOST, USER, PASSWD)
if not check_in(DBNAME, res):
    sql = 'create database %s;' % DBNAME
    oprate_sql(sql, HOST, USER, PASSWD)

　　可以看到，在连接数据库的时候不能指定数据库名称，因为不确定是否存在数据库，所以先查询当前数据库存在哪些数据库名，判断是否存在，不存在则建立数据库名，创建表类似。

　　但是后面也还要操作数据库，比如增删改查，这些都要指定具体数据库名称才可以，那么久不能用上面的函数了，因为数据库名是必填参数，上述函数并没有，代码如下：

def oprate_sql(sql, HOST, USER, PASSWD, DBNAME, charset='utf8', port=3306):#定义操作sql语句的函数，如果是select或show则返回查询内容
    import pymysql
    conn = pymysql.connect(host=HOST, user=USER, password=PASSWD, db=DBNAME, charset=charset)
    cur = conn.cursor()
    cur.execute(sql)
    if sql.strip().startswith('select') or sql.strip().startswith('show'):
        res = cur.fetchall()
    else:
        conn.commit()
        res = 'ok'
    cur.close()
    conn.close()
    return res
#调用：
sql = 'select username from %s' % (USERTB)
res=oprate_sql(sql, HOST, USER, PASSWD,DBNAME)

　　那么问题来了，怎么合并这两个处理sql的函数呢，还记得我们函数有默认参数吧，我们把DBNAME默认为空字符，调用的时候传了DBNAME，那么我们就连接到具体的数据库名，如果调用的时候没有传DBNAME，那么我们就不连接数据库名，代码如下：

def oprate_sql(sql, HOST, USER, PASSWD, DBNAME='', charset='utf8', port=3306):#定义操作sql语句的函数，如果是select或show则返回查询内容
    import pymysql
    if DBNAME:
        conn = pymysql.connect(host=HOST, user=USER, password=PASSWD, db=DBNAME, charset=charset)
    else:
        conn = pymysql.connect(host=HOST, user=USER, password=PASSWD, charset=charset)
    cur = conn.cursor()
    cur.execute(sql)
    if sql.strip().startswith('select') or sql.strip().startswith('show'):
        res = cur.fetchall()
    else:
        conn.commit()
        res = 'ok'
    cur.close()
    conn.close()
    return res
#不传DBNAME
sql='show databases;'
res=oprate_sql(sql, HOST, USER, PASSWD)
#传DBNAME：
sql = 'select username from %s' % (USERTB)
res=oprate_sql(sql, HOST, USER, PASSWD,DBNAME=Table_Name)

　　3.后面学到了类，可以进一步用类封装mysql操作：

 1 import pymysql
 2 
 3 
 4 class MySql(object):
 5     def __init__(self, host, port, user, pwd, dbname="", charset="utf8"):
 6         """初始化数据库信息以及连接"""
 7         self.host = host
 8         self.port = port
 9         self.user = user
10         self.pwd = pwd
11         self.dbname = dbname
12         self.charset = charset
13         self.conn = None  # 给析构函数用的，如果存在链接，则需要关闭链接，没有就不用管
14         self.conn = self.get_conn
15         self.cur = None
16         self.cur = self.get_cur
17 
18     @property
19     def get_conn(self):
20         """根据传参判断是否有传数据库名，然后做不同的链接操作"""
21         try:
22             if self.dbname:
23                 conn = pymysql.connect(host=self.host, port=self.port, user=self.user,
24                                             password=self.pwd, database=self.dbname, charset=self.charset)
25                 return conn
26             else:
27                 conn = pymysql.connect(host=self.host, port=self.port, user=self.user,
28                                             password=self.pwd, charset=self.charset)
29                 return conn
30         except Exception as e:
31             print(e)
32 
33     @property
34     def get_cur(self):
35         """获取游标"""
36         if self.conn:
37             return self.conn.cursor()
38 
39     def exec_sql(self, sql, *args):
40         """
41         执行sql，根据不同的sql语法做不同的操作：
42         查询就返回查询结果列表，其他成功就返回受影响的行数整型，报错返回报错信息字符串
43         *args是为了防止sql注入，自己拼接的sql语法字符串可能被sql注入,
44         这里的用法是sql不需要自己拼接，直接将变量按顺序传进来，pymysql自动拼接，而且避免了sql注入的问题
45         如：
46         sql = "insert into test.student VALUES (19,%s,'English',%s);"
47         res = test_mysql.exec_sql(sql,"yangxga",100)
48         """
49         try:
50             rows = self.cur.execute(sql, args)  # rows记录受影响的行
51             if sql.strip().startswith('select') or sql.strip().startswith('show'):
52                 res = self.cur.fetchall()
53                 res = self.format_res(res)
54                 return res
55             else:
56                 self.conn.commit()  # 非查询语句需要提交才能生效
57                 res = rows
58                 return res
59         except Exception as e:
60             return e
61 
62     @staticmethod
63     def format_res(res):
64         """格式化数据库查找到的结果，如果"""
65         res_lis = []
66         if res:
67             for i in res:
68                 res_lis.append(i)
69         return res_lis
70 
71     def __del__(self):
72         """析构函数，关闭鼠标，断开连接"""
73         if self.cur:
74             self.cur.close()
75         if self.conn:
76             self.conn.close()