python笔记7：接口实现方法

接口基础知识：

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

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

请求方式：get、post方式

请求参数格式：

a. get请求都是通过url?param=xxx¶m1=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、passwd10 '''11 #创建一个服务，把当前这个python文件当做一个服务12 server = flask.Flask(__name__)

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还是POST19 if request.method == 'POST':20 # 获取请求参数是json格式，返回结果是字典21 params = request.json22 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' ;'%username27 # 查询注册的用户是否存在数据库，如果存在，则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可以指定端口，默认端口是500046 #host写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，默认是127.0.0.147 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请求获取请求的参数，返回结果类型是str19 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' ;'%username24 # 查询注册的用户是否存在数据库，如果存在，则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可以指定端口，默认端口是500042 #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.json16 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存入redis20 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可以指定端口，默认端口是500027 #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=sites15 dic = opRedis.get_hashall('sites')16 redisList = []17 for key, value in dic.items():18 redis_dic = {}19 #将字典内元素的类型由bytes转换为str20 k = key.decode()21 v = value.decode()22 #字典redis_dic内结构{'username:k, 'url':v}23 redis_dic['username'] = k24 redis_dic['url'] = v25 redisList.append(redis_dic)26 return jsonify({'code': 200, 'msg': redisList})27 28 if __name__ == '__main__':29 #port可以指定端口，默认端口是500030 #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 print('token',token)17 if name and token:18 #查看数据库，看查询的用户是否存在，若存在则返回用户id19 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可以指定端口，默认端口是500037 #host默认是127.0.0.1,写成0.0.0.0的话，其他人可以访问，代表监听多块网卡上面，38 server.run(debug=True, port=8899, host='0.0.0.0')

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