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)
本文章為轉載內容，我們尊重原作者對文章享有的著作權。如有內容錯誤或侵權問題，歡迎原作者聯繫我們進行內容更正或刪除文章。
