python使用笔记22--mock接口开发

1.mock接口开发

mock是模拟一个接口的意思

为了不阻止测试，开发一个接口，返回你想要的数据，模拟各种场景

需要安装第三方模块flask，flask是web轻量级开发框架

1.1 flask

pip install flask

  1 #mock 模拟一个接口的意思

  2 #http://test.pay.alibab.com/payment

  3 #1、为了不阻碍测试，开发了一个接口，返回你想要的数据，模拟各种场景

  4

  5 import flask

  6 import json

  7 import pymysql

  8 import hashlib

  9 import redis

 10 import time

 11 #flask web轻量级开发框架

 12 server = flask.Flask(__name__)

 13

 14

 15 mysql_info = {'host':'127.0.0.1',

 16               'port':3306,

 17               'user':'root',

 18               'password':'123456',

 19               'autocommit':True,

 20               'db':'db001',

 21               'charset':'utf8'}

 22

 23

 24 #md5加密之后，是固定的，这样的安全性不够高

 25 #可以加盐，salt='xiaoxiaolin_rew43!@'，设置一串复杂的字符串，加强安全性

 26 def md5(s,salt='xiaoxiaolin_rew43!@'):#md5加密

 27     s = str(s+salt)

 28     m = hashlib.md5(s.encode())

 29     return m.hexdigest()

 30

 31 def op_redis(key,value=None,expire=60*60*2):

 32     r = redis.Redis(host="127.0.0.1", port=6379, password="123456&*", decode_responses=True)

 33     if value:#如果传入值不为空，则保存值

 34         r.set(key,value,expire)

 35     else:#value值为空，则获取值

 36         return r.get(key)

 37

 38

 39

 40 def select(sql,oneTag=False):

 41         connect = pymysql.connect(**mysql_info)#**表示将字典转成host='xxx',port=3306

 42         cur = connect.cursor(pymysql.cursors.DictCursor)#获取游标,获取表头

 43         cur.execute(sql)

 44         if oneTag:

 45             result = cur.fetchone()

 46         else:

 47             result = cur.fetchall()#获取查询结果

 48         print(result)

 49         cur.close()

 50         connect.close()

 51         return result

 52

 53

 54 @server.route('/api/payment',methods=['get'])#methods:用来指定请求方式，get/post

 55 def payment():

 56     data = {"code":0,"msg":"支付成功","amount":30000}

 57     return json.dumps(data,ensure_ascii=False)

 58

 59 @server.route('/api/account_data',methods=['get'])

 60 def account_data():

 61     account_name = flask.request.values.get('account_name')#获取请求参数，key-value形式的

 62     #flask.request.json.get()#获取json格式的参数

 63     print(account_name)

 64     result = select("select * from account_info where uname = '%s';" %account_name)

 65     print(result)

 66     data = {'code':0,'msg':'success','data':result}

 67     return json.dumps(data,ensure_ascii=False)

 68

 69 @server.route('/api/account_register',methods=['post'])

 70 def register():#注册

 71     uname = flask.request.values.get('uname')

 72     password = flask.request.values.get('password')

 73     cpwd = flask.request.values.get('cpwd')

 74     select_sql = "select * from account_info where uname = '%s';" % uname

 75     if uname and password and cpwd:#输入参数均不为空

 76         if password != cpwd:

 77             data = {'code': 401, 'msg': '两次输入密码不一致'}

 78         elif select(select_sql):#如果查询结果不为空，则不能注册

 79             data = {'code': 402, 'msg': '输入的账号已存在'}

 80         else:

 81             md5_pwd = md5(password)

 82             insert_sql = "insert into account_info(uname,password) values('%s','%s');"%(uname,md5_pwd)

 83             select(insert_sql)

 84             data = {'code': 200, 'msg': '注册成功'}

 85

 86     else:

 87         data = {'code':400,'msg':'输入参数不能为空'}

 88

 89     return json.dumps(data,ensure_ascii=False)

 90

 91 #登录成功后要返回token，token=username+timestamp md5加密

 92 @server.route('/api/login',methods=['post'])

 93 def login():

 94     uname = flask.request.values.get('uname')

 95     password = flask.request.values.get('password')

 96     if uname and password:

 97         select_sql = "select * from account_info where uname = '%s';" % uname

 98         result = select(select_sql,True)

 99         print(md5(password))

100         if result:

101             if md5(password) == result.get('password'):

102                 token = md5(uname+str(time.time()))#token:uname+timestamp md5加密，保存到redis中，设置有效时间

103                 info = {'username':uname,'id':result.get('id')}

104                 op_redis(token,json.dumps(info))

105                 data = {'code': 200, 'msg': '登录成功','token':token}

106             else:

107                 data = {'code': 401, 'msg': '用户不存在'}

108         else:

109             data = {'code': 403, 'msg': '输入密码错误'}

110     else:

111         data = {'code': 402, 'msg': '账号/密码不能为空'}

112

113     return json.dumps(data,ensure_ascii=False)

114

115 @server.route('/api/payment_v1',methods=['post'])

116 def payment_v1():

117     token = flask.request.values.get('token')

118     amount = flask.request.values.get('amount')

119     if token:#如果token不为空，则去redis中查询用户信息

120         result = op_redis(token)

121         if result:

122             amount = float(amount)

123             result = json.loads(result)

124             userid = result.get('id')

125             update_sql = 'update account_info set balance = balance-%s where id =%s ; '%(amount,userid)

126             print(update_sql)

127             select(update_sql)

128             data = {'code': 0, 'msg': '支付成功'}

129         else:

130             data = {'code': 400, 'msg': '用户未登录'}

131     else:

132         data = {'code':400,'msg':'用户未登录'}

133

134     return  json.dumps(data,ensure_ascii=False)

135

136

137

138 server.run(host='0.0.0.0',port=9999,debug=True)#服务启动只启动一次，服务启动后，不会再执行下面的代码，debug：改代码后不需要重启服务

139 #host='0.0.0.0',这个不写只能在本机访问，如果写上，别人也可以调用你的接口

1.2 fastapi

pip install fastapi

pip install uvicorn

 1 import fastapi

 2 import uvicorn

 3 from pydantic import BaseModel

 4

 5

 6

 7 server = fastapi.FastAPI()#实例化一个接口

 8 class User(BaseModel):

 9     username:str

10     password:str

11     phone:str

12

13

14 @server.post('/user')

15 def user(u:User):#入参是json类型的，要先定义一个类，继承BaseModel

16     print(User)

17     return {'code':0}

18

19 @server.get('/index')

20 def index(id:int,sex:str):

21     if sex=='nan':

22         data = {'id':id,'sex':'男'}

23     else:

24         data = {'id': id, 'sex': '女'}

25

26

27 @server.post('/order')#@:装饰器

28 def order(type:str):

29     if type == 'success':

30         data = {'id':1,'order_amount':10000}

31     else:

32         data = {'id':0,'order_amount':0}

33

34 @server.post('/login')

35 def login(username:str,password:str):

36     if username and password:

37         data = {'code':0,'msg':'登录成功'}

38     else:

39         data = {'code': 0, 'msg': '登录失败'}

40

41

42

43 if __name__ == '__main__':

44     uvicorn.run('fastapi接口开发:server',port=9000,debug=True)#第一个参数是python文件名

2.加密

对称加密：加密和解密用的同样的秘钥

不对称加密：加密和解密用的不同的秘钥

md5加密：md5加密是不可逆转的

1 #md5加密是不可逆转的

2 import hashlib

3 s = '123456'

4 m = hashlib.md5(s.encode())#md5 传入的值是bytes

5 #不管字符串多长，加密后都是32位的值

6 #m = hashlib.sha224(s.encode())

7

8 result = m.hexdigest()#加密

9 print(result)

3.token产生和校验

在登录成功后，都会返回一个token

token是一个唯一的，随机的，且有时效性的值

下面生成token的规则一般是：用户名+timestamp 再md5加密

token保存在redis中，以便于后面接口校验token

3.1.登录接口--生成token

 1 #登录成功后要返回token，token=username+timestamp md5加密

 2 @server.route('/api/login',methods=['post'])

 3 def login():

 4     uname = flask.request.values.get('uname')

 5     password = flask.request.values.get('password')

 6     if uname and password:

 7         select_sql = "select * from account_info where uname = '%s';" % uname

 8         result = select(select_sql,True)

 9         print(md5(password))

10         if result:

11             if md5(password) == result.get('password'):

12                 token = md5(uname+str(time.time()))#token:uname+timestamp md5加密，保存到redis中，设置有效时间

13                 info = {'username':uname,'id':result.get('id')}

14                 op_redis(token,json.dumps(info))

15                 data = {'code': 200, 'msg': '登录成功','token':token}

16             else:

17                 data = {'code': 401, 'msg': '用户不存在'}

18         else:

19             data = {'code': 403, 'msg': '输入密码错误'}

20     else:

21         data = {'code': 402, 'msg': '账号/密码不能为空'}

22

23     return json.dumps(data,ensure_ascii=False)

3.2.支付接口--校验token

 1 @server.route('/api/payment_v1',methods=['post'])

 2 def payment_v1():

 3     token = flask.request.values.get('token')

 4     amount = flask.request.values.get('amount')

 5     if token:#如果token不为空，则去redis中查询用户信息

 6         result = op_redis(token)

 7         if result:

 8             amount = float(amount)

 9             result = json.loads(result)

10             userid = result.get('id')

11             update_sql = 'update account_info set balance = balance-%s where id =%s ; '%(amount,userid)

12             print(update_sql)

13             select(update_sql)

14             data = {'code': 0, 'msg': '支付成功'}

15         else:

16             data = {'code': 400, 'msg': '用户未登录'}

17     else:

18         data = {'code':400,'msg':'用户未登录'}

19

20     return  json.dumps(data,ensure_ascii=False)

巴特西