1 Star 1 Fork 3

测试开发学苑/user_api_server

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
该仓库未声明开源许可证文件(LICENSE),使用请关注具体项目描述及其代码上游依赖。
克隆/下载
api_server_plus.py 5.53 KB
一键复制 编辑 原始数据 按行查看 历史
狂师 提交于 2021-10-22 10:45 +08:00 . add:添加项目源码文件
import hashlib
import hmac
import json
import random
import string
from functools import wraps
from flask import Flask, make_response, request
from faker import Faker
SECRET_KEY = "mikezhou"
app = Flask(__name__)
users_dict = {}
token_dict = {}
fake = Faker(locale='zh_CN')
def gen_random_string(str_len):
""" generate random string with specified length
"""
return ''.join(
random.choice(string.ascii_letters + string.digits) for _ in range(str_len))
def get_sign(*args):
content = ''.join(args).encode('ascii')
sign_key = SECRET_KEY.encode('ascii')
sign = hmac.new(sign_key, content, hashlib.sha1).hexdigest()
return sign
def gen_md5(*args):
return hashlib.md5("".join(args).encode('utf-8')).hexdigest()
def validate_request(func):
@wraps(func)
def wrapper(*args, **kwargs):
device_sn = request.headers.get('device_sn', "")
token = request.headers.get('token', "")
if not device_sn or not token:
result = {
'success': False,
'msg': "device_sn or token is null."
}
response = make_response(json.dumps(result), 401)
response.headers["Content-Type"] = "application/json"
return response
if token_dict.get(device_sn) != token:
result = {
'success': False,
'msg': "Authorization failed!"
}
response = make_response(json.dumps(result), 403)
response.headers["Content-Type"] = "application/json"
return response
return func(*args, **kwargs)
return wrapper
@app.route('/')
def index():
return "Hello World!"
@app.route('/api/get-token', methods=['POST'])
def get_token():
device_sn = request.headers.get('device_sn', "")
os_platform = request.headers.get('os_platform', "")
app_version = request.headers.get('app_version', "")
data = request.get_json()
sign = data.get('sign', "")
expected_sign = get_sign(device_sn, os_platform, app_version)
if expected_sign != sign:
result = {
'success': False,
'msg': "Authorization failed!"
}
response = make_response(json.dumps(result), 403)
else:
token = gen_random_string(16)
token_dict[device_sn] = token
result = {
'success': True,
'token': token
}
response = make_response(json.dumps(result))
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/users')
@validate_request
def get_users():
users_list = [user for uid, user in users_dict.items()]
users = {
'success': True,
'count': len(users_list),
'items': users_list
}
response = make_response(json.dumps(users))
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/reset-all')
@validate_request
def clear_users():
users_dict.clear()
result = {
'success': True
}
response = make_response(json.dumps(result))
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/users/<int:uid>', methods=['POST'])
@validate_request
def create_user(uid):
user = request.get_json()
if uid not in users_dict:
result = {
'success': True,
'msg': "user created successfully."
}
status_code = 201
users_dict[uid] = user
else:
result = {
'success': False,
'msg': "user already existed."
}
status_code = 500
response = make_response(json.dumps(result), status_code)
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/users/<int:uid>')
@validate_request
def get_user(uid):
user = users_dict.get(uid, {})
info={
'address': fake.address(),
'name': fake.name(),
'ssn': fake.ssn(),
'company': fake.company(),
'phone': fake.phone_number(),
'imageurl': fake.image_url(),
'url': fake.url(),
}
if user:
result = {
'success': 'get user success',
'data': user,
'info':info,
}
status_code = 200
else:
result = {
'success': 'get user failed',
'data': user
}
status_code = 404
response = make_response(json.dumps(result), status_code)
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/users/<int:uid>', methods=['PUT'])
@validate_request
def update_user(uid):
user = users_dict.get(uid, {})
if user:
user = request.get_json()
success = 'update success'
status_code = 200
users_dict[uid] = user
else:
success = 'update failed'
status_code = 404
result = {
'success': success,
'data': user
}
response = make_response(json.dumps(result), status_code)
response.headers["Content-Type"] = "application/json"
return response
@app.route('/api/users/<int:uid>', methods=['DELETE'])
@validate_request
def delete_user(uid):
user = users_dict.pop(uid, {})
if user:
success = 'delete success'
status_code = 200
else:
success = 'delete failed'
status_code = 404
result = {
'success': success,
'data': user
}
response = make_response(json.dumps(result), status_code)
response.headers["Content-Type"] = "application/json"
return response
if __name__ == '__main__':
app.run()
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Python
1
https://gitee.com/mikezhou_talk/user-api-server.git
git@gitee.com:mikezhou_talk/user-api-server.git
mikezhou_talk
user-api-server
user_api_server
master

搜索帮助