python笔记7：接口实现方法

接口基础知识：

简单说下接口测试，现在常用的2种接口就是http api和rpc协议的接口，今天主要说：http api接口是走http协议通过路径来区分调用的方法，请求报文格式都是key-value形式，返回报文一般是json串；

接口协议：http、webservice、rpc等。

请求方式：get、post方式

请求参数格式：

　　a. get请求都是通过url?param=xxx&param1=xxx

　　b. post请求的请求参数常用类型有：application/json、application/x-www-form-urlencoded、multipart/form-data、text/html等。

还需要知道接口的url、参数类型、返回结果的数据格式、了解接口是否有header、cookie等信息。

接口的实现：

请求方式-get，接口的写法：

 1 import flask
 2 from flask import request
 3 from flask import jsonify
 4 import tools
 5 import OP_db
 6 import settings
 7 '''
 8 flask： web框架，可以通过flask提供的装饰器@server.route()将普通函数转换为服务
 9 登录接口，需要传url、username、passwd
10 '''
11 #创建一个服务，把当前这个python文件当做一个服务
12 server = flask.Flask(__name__)
13 #浏览器访问接口时，返回的字符串显示为utf-8的编码，不是中文，可以使用以下代码解决13 server.config['JSON_AS_ASCII'] = False
14 
15 # @server.route()可以将普通函数转变为服务  登录接口的路径、请求方式
16 @server.route('/login', methods=['get'])
17 def login():
18     # 获取通过url请求传参的数据
19     username = request.values.get('name')
20     # 获取url请求传的密码，明文
21     pwd = request.values.get('pwd')
22     # 判断用户名、密码都不为空，如果不传用户名、密码则username和pwd为None
23     if username and pwd:
24         # 获取加密后的密码
25         password = tools.md5_pwd(pwd)
26         #执行sql，如果查询的username和password不为空，说明数据库存在admin的账号
27         sql = 'select name,password from test where name= "%s" and password= "%s";' %(username, password)
28         # 从数据查询结果后，res返回是元组
29         res = OP_db.getconn(
30             host=settings.mysql_info['host'],
31             user=settings.mysql_info['user'],
32             passwd=settings.mysql_info['pwd'],
33             db=settings.mysql_info['db'],
34             port=settings.mysql_info['port'],
35             sql=sql
36         )
37         if res:      #res的结果不为空，说明找到了username=admin的用户，且password为加密前的123456
38             resu = {'code': 200, 'message': '登录成功'}
39             return jsonify(resu) #将字典转换为json串, json是字符串
40         else:
41             resu = {'code': -1, 'message': '账号/密码错误'}
42             return jsonify(resu)
43     else:
44         res = {'code': 999, 'message': '必填参数未填写'}
45         return jsonify(res)
46 
47 if __name__ == '__main__':
48       server.run(debug=True, port=8888, host=0.0.0.0)  #指定端口、host,0.0.0.0代表不管几个网卡，任何ip都可以访问

md5加密、数据库mysql的操作详见我的其他博客~~~~~

get访问接口：

项目启动后，接口的地址是：http://127.0.0.1:5000/，默认端口是5000。

打开浏览器，输入urlhttp://127.0.0.1:5000/xxx?name=xxx&pwd=123456，后面跟上接口的地址login，参数跟url直接使用？相连，每个请求参数直接使用&相连。请求成功，则返回{'code': 200, 'message': '登录成功'}。

请求方式-post，接口的写法：

 1 import flask
 2 from flask import jsonify
 3 from flask import request
 4 from conf import opMysql
 5 from conf import md5_create
 6 '''
 7 注册接口：
 8 post请求，请求参数入参类型json
 9 {
10     "username":"aaa",
11     "pwd":"123456",
12     "c_pwd":"123456"
13 }
14 '''
15 server = flask.Flask(__name__)
16 @server.route('/register', methods=['GET', 'POST'])
17 def registerPost():
18     #判断接口的请求方式是GET还是POST
19     if request.method == 'POST':
20         # 获取请求参数是json格式，返回结果是字典
21         params = request.json
22         username = params.get('username')
23         pwd = params.get('pwd')
24         confirmpwd = params.get('confirmpwd')
25         if username and pwd and confirmpwd:  # 判断输入的用户名、密码、确认密码都不为空
26             select_sql = 'select username from lhldemo  where username = "%s" ;'%username
27             # 查询注册的用户是否存在数据库，如果存在，则username不为空，否则username为空
28             res_mysql = opMysql.op_select(select_sql)
29             if res_mysql:
30                 return jsonify({"code": 999, "mesg": "用户已注册"})
31             else:
32                 if pwd == confirmpwd:  # 判断pwd和confirmpwd一致
33                     new_pwd = md5_create.md5_test(pwd)  # 加密后的密码
34                     insert_sql = 'insert into lhldemo(username,password) values("%s", "%s") ;' % (username, new_pwd)
35                     opMysql.op_insert(insert_sql)
36                     return jsonify({"code": 200, "msg": "注册成功"})
37                 else:
38                     return jsonify({"code":998, "msg":"密码不一样"})
39         else:
40             return jsonify({"code": 504, "msg": "必填项不能为空"})
41     else:
42         return jsonify({"code": 201, "msg": "请求方式不正确"})
43 
44 if __name__ == '__main__':
45     #port可以指定端口，默认端口是5000
46     #host写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，默认是127.0.0.1
47     server.run(debug=True, port=8899, host='0.0.0.0')

post访问接口：

项目启动后，接口的地址是：http://127.0.0.1:5000/，默认端口是5000。

打开浏览器，输入urlhttp://127.0.0.1:5000/xxx，后面跟上接口的地址register，参数使用postman或jmeter进行请求，参数类型是json。请求成功，则返回{'code': 200, 'message': '登录成功'}。

请求方式-get、post都可以访问，写法如下：

 1 import flask
 2 from flask import jsonify
 3 from flask import request
 4 from conf import opMysql
 5 from conf import md5_create
 6 '''
 7 注册接口：
 8 post请求，请求参数入参类型json
 9 {
10     "username":"aaa",
11     "pwd":"123456",
12     "c_pwd":"123456"
13 }
14 '''
15 server = flask.Flask(__name__)
16 @server.route('/register', methods=['get', 'post'])
17 def registerPost():
18     #post请求获取请求的参数，返回结果类型是str
19     username = request.values.get('username')
20     pwd = request.values.get('pwd')
21     confirmpwd = request.values.get('confirmpwd')
22     if username and pwd and confirmpwd:  # 判断输入的用户名、密码、确认密码都不为空
23         select_sql = 'select username from lhldemo  where username = "%s" ;'%username
24         # 查询注册的用户是否存在数据库，如果存在，则username不为空，否则username为空
25         res_mysql = opMysql.op_select(select_sql)
26         if res_mysql:
27             return jsonify({"code": 999, "mesg": "用户已注册"})
28         else:
29             if pwd == confirmpwd:  # 判断pwd和confirmpwd一致
30                 new_pwd = md5_create.md5_test(pwd)  # 加密后的密码
31                 insert_sql = 'insert into lhldemo(username,password) values("%s", "%s") ;' % (username, new_pwd)
32                 opMysql.op_insert(insert_sql)
33                 return jsonify({"code": 200, "msg": "注册成功"})
34             else:
35                 return jsonify({"code": 998, "msg": "密码不一样"})
36     else:
37         return jsonify({"code": 504, "msg": "必填项不能为空"})
38 
39 
40 if __name__ == '__main__':
41     #port可以指定端口，默认端口是5000
42     #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
43     server.run(debug=True, port=8899, host='0.0.0.0')

可以通过以下2种方式进行post请求，一种如下：

通过url拼接参数：

第二种访问方式：通过key-value方式进行访问：

redis相关操作，添加hash类型的值到redis内，接口实现如下：

 1 import flask
 2 from flask import jsonify
 3 from conf import opRedis
 4 from flask import request
 5 '''
 6 redis添加数据，存入数据的类型是hash类型，格式如下：
 7 post请求，请求参数入参类型json
 8 {name:{"key":"value"}}
 9 {"username":"url"}
10 '''
11 server = flask.Flask(__name__)
12 @server.route('/set_sties', methods =['post'])
13 def set_sties():
14     # 获取url请求参数，返回结果是字典{"username":"byz","url":"http://www.baidu.com"}
15     res_dic = request.json
16     if res_dic.get('username') and res_dic.get('url'):
17         username = res_dic.get('username')
18         url = res_dic.get('url')
19         #调用redis的hset方法，将username、url存入redis
20         opRedis.get_hashall('sites', username, url)
21         return jsonify({"code":20})
22     else:
23         return jsonify({"code": 204, "msg": "必填项不能为空"})
24 
25 if __name__ == '__main__':
26     #port可以指定端口，默认端口是5000
27     #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
28     server.run(debug=True, port=8899, host='0.0.0.0')

hash类型结构如下：

{name:{key,value}}，接口访问成功后，redis内数据存储结构如下：

redis添加完数据后，读取redis内的数据，接口实现如下：

 1 import flask
 2 from flask import jsonify
 3 from conf import opRedis
 4 from flask import request
 5 '''
 6 读取redis内的数据，redis数据存储类型是hash类型，格式如下
 7 {name:{"key":"value"}}
 8 思路： 1.通过redis的hgetall(name)方法读取redis所有数据，返回结果类型是字典
 9        2. 循环字典内容，将元素类型转换为str，并将结果存放到字典内
10 '''
11 server = flask.Flask(__name__)
12 @server.route('/get_sties', methods =['get', 'post'])
13 def get_sties():
14     #获取redis内所有的数据信息，返回结果类型是字典，里面元素是bytes类型，name=sites
15     dic = opRedis.get_hashall('sites')
16     redisList = []
17     for key, value in dic.items():
18         redis_dic = {}
19         #将字典内元素的类型由bytes转换为str
20         k = key.decode()
21         v = value.decode()
22         #字典redis_dic内结构{"username:k, "url":v}
23         redis_dic['username'] = k
24         redis_dic['url'] = v
25         redisList.append(redis_dic)
26     return jsonify({"code": 200, "msg": redisList})
27 
28 if __name__ == '__main__':
29     #port可以指定端口，默认端口是5000
30     #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
31     server.run(debug=True, port=8899, host='0.0.0.0')

通过postman方法接口，返回数据如下：

查询用户，需要传token值，实现方法如下：

登录接口：

 1 import flask
 2 from flask import jsonify
 3 from conf import opRedis
 4 from conf import opMysql
 5 from conf import md5_create
 6 from flask import request
 7 import time
 8 '''
 9 登录接口，需要传用户名、密码，通过查询数据库判断用户是否登录成功，若登录成功则将用户名和token存入redis内
10 '''
11 server = flask.Flask(__name__)
12 @server.route('/login', methods=['get','post'])
13 def set_cookies():
14     name = request.values.get('username')
15     pwd = request.values.get('pwd')
16     if name and pwd:
17         #加密后的密码
18         new_pwd = md5_create.md5_test(pwd)
19         sql = 'select username,password from lhldemo where username="%s" and password="%s" ; ' % (name, new_pwd)
20         res_sql = opMysql.op_select(sql)
21         if res_sql:
22             token = name + time.strftime('%Y%m%d%H%M%S')
23             new_token = md5_create.md5_test(token)
24             #用户登录成功后，将name和token存入redis，存入数据类型是hash类型
25             opRedis.get_hashall('user', name, new_token)
26             return jsonify({"code": 200})
27         else:
28             return jsonify({"code": 204})
29     else:
30         return jsonify({"code": 304})

查询用户，需要传用户名和token值，实现方法如下：

 1 import flask
 2 from flask import jsonify
 3 from conf import opRedis
 4 from conf import opMysql
 5 from conf import md5_create
 6 from flask import request
 7 import time
 8 '''
 9 登录接口，需要传用户名、密码，通过查询数据库判断用户是否登录成功，若登录成功则将用户名和token存入redis内
10 '''
11 server = flask.Flask(__name__)
12 @server.route('/search_user', methods=['get','post'])
13 def set_cookies():
14     name = request.values.get('username')
15     token = request.values.get('token')
16     17     if name and token:
18         #查看数据库，看查询的用户是否存在，若存在则返回用户id
19         sql = 'select id from lhldemo where username="%s" ; ' % (name)
20         res_sql = opMysql.op_select(sql)
21         if res_sql:
22             #从redis中获取user下的用户名对应的token值
23             res_token = opRedis.getRedis('user:'+name)26             if res_token == token:
27                 return jsonify({"msg": "用户id", "id": res_sql})
28             else:
29                 return jsonify({"msg": "token错误"})
30         else:
31             return jsonify({"code": "用户不存在"})
32     else:
33         return jsonify({"code": "必填项不能为空"})
34 
35 if __name__ == '__main__':
36     #port可以指定端口，默认端口是5000
37     #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，
38     server.run(debug=True, port=8899, host='0.0.0.0')

文件的上传、下载接口实现：

上传接口：

 1 import flask
 2 from flask import request
 3 from flask import jsonify
 4 import time
 5 
 6 server = flask.Flask(__name__)
 7 @server.route('/upload', methods=['post'])
 8 def upload():
 9     #获取请求参数，传参类型是file,返回结果类型是一个对象：<class 'werkzeug.datastructures.FileStorage'> <FileStorage: '新建文本文档.txt' ('text/plain')>
10     file = request.files.get('file_name', None)
11     #判断file是否为空，若为空则没有上传文件
12     if file:
13         cur_time = time.strftime('%Y%m%d%H%M%S')  #获取当前时间
14         upload_name = file.filename               #获取上传文件的名称
15         new_file_name = cur_time+upload_name      #给文件重命名，防止有重复文件覆盖
16         # 保存文件,指定文件上传的路径
17         file.save('E:/python_workspace/base-code/day8/test_requests/'+new_file_name)
18         return jsonify({"code": "ok"})
19     else:
20         return jsonify({"code": "请上传文件"})
21 
22 server.run(debug=True)

通过postman访问接口：

下载接口：

 1 @server.route('/get_file', methods = ['get'])
 2 def get_file():
 3     #获取请求参数,返回结果类型是str
 4     filename = request.values.get('filename', None)
 5     if filename:
 6         #判断请求的参数，也就是要下载的文件，在当前目录下是否是一个文件，而不是文件夹
 7         if os.path.isfile('E:/python_workspace/base-code/'+filename):
 8             #返回要下载的文件
 9             return send_from_directory('.', filename, as_attachment=True)
10         else:
11             return jsonify({"msg": "文件不存在!"})
12     else:
13         return jsonify({'msg': '文件名不能为空'})
14 
15 if __name__ == '__main__':
16     server.run(debug=True)

以上就是工作中常用的一些接口场景，测试支付相关接口、或者第三方接口时，可以自己mock接口返回假数据操作~~~~