使用|使用 Flask-RESTful 设计 RESTful API
一、简述
RESTful API 的功能已经实现了，这里我只想讲解一下代码，实现步骤就不说了，不然太耗时。首先我要讲解一下框架结构，说清楚每个文件做什么用的（其实之前我写的一篇文章里已经说明过了，有兴趣的可以回去再看一下）。然后讲解一下代码细节，和功能是如何实现的。最后通过终端验证一下。
二、框架说明
1. 总体结构展示
【使用|使用 Flask-RESTful 设计 RESTful API】我们来看一下整理的结构。
文章图片
整体架构.png 三、细节代码分析
1. 依赖的包
其实我这里使用的是pip安装。
(venv303) [root@test01 ~]# pip install flask flask-script flask-sqlalchemy flask-migrate flask_restful pymysql flask-httpauth
我们也可以看下requirements.txt文件。
(venv303) [root@test01 pycharm_project_486]# pip freeze >requirements.txt
(venv303) [root@test01 pycharm_project_486]# cat requirements.txt
alembic==0.9.9
aniso8601==3.0.0
click==6.7
Flask==0.12.2
Flask-HTTPAuth==3.2.3
Flask-Migrate==2.1.1
Flask-RESTful==0.3.6
Flask-Script==2.0.6
Flask-SQLAlchemy==2.3.2
itsdangerous==0.24
Jinja2==2.10
Mako==1.0.7
MarkupSafe==1.0
PyMySQL==0.8.0
python-dateutil==2.7.2
python-editor==1.0.3
pytz==2018.3
six==1.11.0
SQLAlchemy==1.2.5
Werkzeug==0.14.1
2. 数据库配置文件 主要用来连接数据使用的，这里我们可以创建多个database，以便在不同的环境中使用，开发环境和线上环境本质上的不同，就在于数据嘛。
- config.py
class Config:
SECRET_KEY = 'hard to guess string'
SQLALCHEMY_COMMIT_ON_TEARDOWN = True
SQLALCHEMY_TRACK_MODIFICATIONS = False@staticmethod
def init_app(app):
passclass MySQLConfig:
MYSQL_USERNAME = 'root'
MYSQL_PASSWORD = '123456'
MYSQL_HOST = '192.168.1.30'class DevelopmentConfig(Config):
DEBUG = True
database = 'mysql_dev'
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{}:{}@{}/{}'.format(MySQLConfig.MYSQL_USERNAME,
MySQLConfig.MYSQL_PASSWORD,
MySQLConfig.MYSQL_HOST, database)class TestingConfig(Config):
TESTING = True
database = 'mysql_test'
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{}:{}@{}/{}'.format(MySQLConfig.MYSQL_USERNAME,
MySQLConfig.MYSQL_PASSWORD,
MySQLConfig.MYSQL_HOST, database)class ProductionConfig(Config):
database = 'mysql_product'
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{}:{}@{}/{}'.format(MySQLConfig.MYSQL_USERNAME,
MySQLConfig.MYSQL_PASSWORD,
MySQLConfig.MYSQL_HOST, database)config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,'default': DevelopmentConfig
}3. 管理文件 主要用于启动和管理程序，例如我们可以给这个程序定义端口号，是否是debug模式，是否自动reload（就是更改完代码之后，自动生效，不需要再重启程序）等等。
- manage.py
from app import create_app, db
from flask_script import Server, Manager, Shell
from app.models.pxeinfo import PxeInfo
from app.models.user import Userapp = create_app('default')
manager = Manager(app=app)def make_shell_context():
return dict(app=app, db=db, User=User, PxeInfo=PxeInfo)manager.add_command('runserver', Server(host='192.168.1.30', port=80, use_debugger=True, use_reloader=True))
manager.add_command('shell', Shell(make_context=make_shell_context))if __name__ == '__main__':
manager.run(default_command='runserver')# 这里可以创建shell模式，在shell模式下可以使用命令删除或创建数据库
# 删除的命令是：db.drop_all()，创建的命令是：db.create_all()
# 创建和删除哪些表需要提前将ORM模型引入进来（就是加到make_shell_context函数里）
# manager.run(default_command='shell')4. 主程序的
__init__.py
__init__.py 文件的作用是将文件夹变为一个Python模块,Python 中的每个模块的包中，都有__init__.py 文件。通常
__init__.py 文件为空，但是我们还可以为它增加其他的功能。我们在导入一个包时，实际上是导入了它的__init__.py文件。这样我们可以在__init__.py文件中批量导入我们所需要的模块，而不再需要一个一个的导入（例如api_1_0文件夹下的__init__.py，将其它文件import进来）。- app文件夹下的
__init__.py，使用了工厂模式，创建app实例。这样可以创建多个，并且易于被manage.py维护
from flask import Flask
from config import config
from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config[config_name])
config[config_name].init_app(app=app)
db.init_app(app=app)from .main import main as main_blueprint
app.register_blueprint(main_blueprint)
from .api_1_0 import api_1_0 as api_blueprint
app.register_blueprint(api_blueprint)return app
5. api蓝本 从上面的app文件夹下的
__init__py，我们可以看到app.register_blueprint(api_blueprint)。注册了蓝本的api，接下来我们在api_1_0的文件夹下创建蓝本- api_1_0文件夹下的
__init__.py（这个文件的含义我上面已经说过了）
from flask import Blueprintapi_1_0 = Blueprint('api_1_0', __name__, url_prefix='/api')from . import api_pxe_info, api_user, errors, api_auth
- api_1_0文件夹下的
api_user.py
import timefrom app import db
from flask_restful import Api, Resource
from flask import jsonify, requestfrom app.api_1_0 import api_1_0
from app.models.user import User
from app.api_1_0.api_auth import auth, generate_auth_token, verify_auth_tokenapi_user = Api(api_1_0)class UserAddApi(Resource):
# 添加用户，要求验证
@auth.login_required
def post(self):
user_info = request.get_json()
try:
u = User(username=user_info['username'])
u.password = user_info['password']
db.session.add(u)
db.session.commit()
except:
print("{} User add: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
db.session.rollback()
return False
else:
print("{} User add: {} success...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return True
finally:
db.session.close()class UserVerifyApi(Resource):
# 根据传过来的账号密码，返回验证结果。
@auth.login_required
def post(self):
user_info = request.get_json()
try:
u = User.query.filter_by(username=user_info['username']).first()
if u is None or u.verify_password(user_info['password']) is False:
print("{} User query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return False
except:
print("{} User query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return False
else:
print("{} User query: {} success...".format(time.strftime("%Y-%m-%d %H:%M:%S"), user_info['username']))
return True
finally:
db.session.close()class UserToken(Resource):
# 返回一个token，默认是1个小时有限的token
@auth.login_required
def get(self):
token = generate_auth_token(expiration=3600)
return jsonify({'token': token.decode('ascii')})api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')
api_user.add_resource(UserVerifyApi, '/userverify', endpoint='userverify')
api_user.add_resource(UserToken, '/usertoken', endpoint='usertoken')
- api_1_0文件夹下的
api_pxe_info.py
import timefrom app import db
from ..api_1_0 import api_1_0
from flask_restful import Api, Resource
from flask import jsonify, request
from app.models.pxeinfo import PxeInfo
from app.api_1_0.api_auth import authapi_pxe_info = Api(api_1_0)class TestApi(Resource):
def get(self):
return jsonify({'test_api': 'api is ok'})class PxeInfoApi(Resource):
# 添加信息
@auth.login_required
def post(self):
pxe_info = request.get_json()
print(pxe_info)
print(type(pxe_info))
try:
pxe = PxeInfo(sn=pxe_info['sn'], pxe_ip=pxe_info['pxe_ip'], ilo_ip=pxe_info['ilo_ip'],
mac1=pxe_info['mac1'], mac2=pxe_info['mac2'], sw_name1=pxe_info['sw_name1'],
sw_name2=['sw_name2'], sw_port1=pxe_info['sw_port1'], sw_port2=pxe_info['sw_port2'])
db.session.add(pxe)
db.session.commit()
except:
print("{} PxeInfo add: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), pxe_info['sn']))
db.session.rollback()
return False
else:
print("{} PxeInfo add: {} success...".format(time.strftime("%Y-%m-%d %H:%M:%S"), pxe_info['sn']))
return True
finally:
db.session.close()# 根据GET方式传过来的sn值，查询结果
@auth.login_required
def get(self):
s = request.args.get('sn')
try:
pxe_info = PxeInfo.query.filter_by(sn=s).order_by(PxeInfo.id.desc()).first()
if pxe_info is None:
print("{} PxeInfo query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), pxe_info['sn']))
return False
return pxe_info.to_json()
except:
print("{} PxeInfo query: {} failure...".format(time.strftime("%Y-%m-%d %H:%M:%S"), pxe_info['sn']))
return False
finally:
db.session.close()api_pxe_info.add_resource(TestApi, '/test_api', endpoint='test_api')
api_pxe_info.add_resource(PxeInfoApi, '/pxeinfo', endpoint='pxeinfo')
- api_1_0文件夹下的
api_auth.py
from flask_httpauth import HTTPBasicAuth
from flask import jsonify, app
from itsdangerous import SignatureExpired, BadSignature
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from config import Configfrom app.models.user import Userauth = HTTPBasicAuth()# 请求api接口数据的时候，-u 后面输入的账号密码不正确，返回该值
@auth.error_handler
def unauthorized():
error_info = '{}'.format("Invalid credentials")
print(error_info)
response = jsonify({'error': error_info})
response.status_code = 403
return response# 这个是第一次使用账号密码做验证的时候使用的函数
# 后来发现用token方式访问api更安全，所以就把之前的这个函数注释掉了
# @auth.verify_password
# def verify_password(username, password):
#user = User.query.filter_by(username=username).first()
#if not user or not user.verify_password(password):
#return False
#return True# 只是一个辅助函数，当传入用户名密码的时候，查询数据库中是否有这条记录
# 并且密码也正确，则返回为Ture
def verify_password_for_token(username, password):
user = User.query.filter_by(username=username).first()
if not user or not user.verify_password(password):
return False
return True# 验证 token 和 用户名密码
# 默认传的用户名密码的格式，例如用户名叫liuxin，密码是123456 在shell里加入 -u username:password
# 先验证用户名，首先假想是token，解密，查询是否有这么个用户存在，如果有返回True
# 如果用户名，那么上面解密这个名字，也肯定解密不出来，所以得出来的user是None
# 那么接下来就通过用户名密码的方式验证
# 需要注意的是，传入token的方式与传账号密码的方式一样，别忘记后面加一个冒号:
# url中加入@auth.login_required修饰符，会默认调用此函数
@auth.verify_password
def verify_password(username_or_token, password):
# first try to authenticate by token
user = verify_auth_token(username_or_token)
if user is None:
# try to authenticate with username/password
return verify_password_for_token(username=username_or_token, password=password)
return True# 定义一个产生token的方法
def generate_auth_token(expiration=36000):
# 注意这里的Serializer是这么导入的
# from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
s = Serializer(secret_key="tiantiankaixin", expires_in=expiration)
# print(s.dumps({'id': 1}))
# 返回第一个用户，这里我就将数据库里的id=1的用户作为token的加密用户
return s.dumps({'id': 1})# 解密token，因为上面加密的是 id=1 的用户，所以解密出来的用户也是 id=1 的用户
# 所以data的数值应该是 {'id': 1}
def verify_auth_token(token):
s = Serializer("tiantiankaixin")
try:
data = https://www.it610.com/article/s.loads(token)
except SignatureExpired:
return None# valid token, but expired
except BadSignature:
return None# invalid token
user = User.query.get(data['id'])
return user
- api_1_0文件夹下的
errors.py
from . import api_1_0
from flask import jsonify@api_1_0.app_errorhandler(404)
def not_found(e):
print(e)
error_info = '{}'.format(e)
response = jsonify({'error': error_info})
response.status_code = 404
return response@api_1_0.app_errorhandler(403)
def forbidden(e):
print(e)
error_info = '{}'.format(e)
response = jsonify({'error': error_info})
response.status_code = 403
return response
6. ORM模型 有些书上在模型中创建了很多函数，例如增删改查的操作都写到了这个模型类中。个人感觉不太好，虽然使用起来方便了，但是看起来给人的感觉太凌乱了。如果需要增删改查，可以再专门写一个操作的类。
- models文件夹下的 user.py
from app import db
from werkzeug.security import generate_password_hash, check_password_hashclass User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, index=True)
password_hash = db.Column(db.String(128))# 定义一个属性，默认是读取的操作，这里报错，意思是不可读
@property
def password(self):
raise AttributeError('password is not readable attribute')# 定义上面那个password属性的可写属性，这里默认换算成哈希值，然后保存下来
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)# 校验传入的密码和哈希值是否是一对儿
def verify_password(self, password):
return check_password_hash(self.password_hash, password)def __repr__(self):
return "".format(self.username)
- models文件夹下的 pxeinfo.py
import datetime
from flask import jsonify
from app import dbclass PxeInfo(db.Model):
__tablename__ = 'PxeInfo'
id = db.Column(db.Integer, primary_key=True)
sn = db.Column(db.String(64), index=True)
pxe_ip = db.Column(db.String(64))
ilo_ip = db.Column(db.String(64))
mac1 = db.Column(db.String(64))
mac2 = db.Column(db.String(64))
sw_name1 = db.Column(db.String(64))
sw_name2 = db.Column(db.String(64))
sw_port1 = db.Column(db.String(64))
sw_port2 = db.Column(db.String(64))
info_time = db.Column(db.DateTime)def __init__(self, sn, pxe_ip, ilo_ip, mac1, mac2, sw_name1, sw_name2, sw_port1, sw_port2):
self.sn = sn
self.pxe_ip = pxe_ip
self.ilo_ip = ilo_ip
self.mac1 = mac1
self.mac2 = mac2
self.sw_name1 = sw_name1
self.sw_name2 = sw_name2
self.sw_port1 = sw_port1
self.sw_port2 = sw_port2
self.info_time = datetime.datetime.now()def to_json(self):
j = jsonify({'id': self.id, 'sn': self.sn, 'pxe_ip': self.pxe_ip, 'ilo_ip': self.ilo_ip, 'mac1': self.mac1,
'mac2': self.mac2, 'sw_name1': self.sw_name1, 'sw_name2': self.sw_name2, 'sw_port1': self.sw_port1,
'sw_port2': self.sw_port2, 'info_time': self.info_time})
return jdef __repr__(self):
return "".format(self.sn)
7、数据库的操作
可以在manage.py 文件加入shell参数创建或删除数据库中的表，但是每次都要输命令，所以我写了一个文件，会自动初始化文件
- db文件夹下的init_db.py
from app import create_appdef init_db(mysql_db='default'):
from app.models.pxeinfo import PxeInfo
from app.models.user import User
from app import db
app = create_app(mysql_db)
app.app_context().push()
db.drop_all()
db.create_all()
db.session.commit()init_db()
四、验证
1. 初始化数据库 db文件夹下的init_db.py，创建相应的表，结果如下
文章图片
image.png 2. 添加用户 首先启动程序，然后执行
ssh://root@192.168.1.30:22/root/test/venv/venv303/bin/python -u /tmp/pycharm_project_486/manage.py
* Running on http://192.168.1.30:80/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 252-250-956
添加账号失败。
[root@test01 ~]# curl -H "Content-Type:application/json" -X POST --data '{"username":"liuxin","password":"tiantiankaixin"}' http://192.168.1.30/api/useradd
{
"error": "Invalid credentials"
}
原因是users数据库中没有任何数据，而在添加用户的时候需要账号密码验证，所以我们暂时先把验证方式注释掉
class UserAddApi(Resource):
# 添加用户，要求验证
# @auth.login_required
保存，因为manage.py中添加了use_reloader=True，所以无需手动重启服务
ssh://root@192.168.1.30:22/root/test/venv/venv303/bin/python -u /tmp/pycharm_project_486/manage.py
* Running on http://192.168.1.30:80/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 252-250-956
Invalid credentials
192.168.1.30 - - [30/Mar/2018 21:16:51] "POST /api/useradd HTTP/1.1" 403 -
* Detected change in '/tmp/pycharm_project_486/app/api_1_0/api_user.py', reloading
* Restarting with stat
* Debugger is active!
* Debugger PIN: 252-250-956
再次尝试添加账号
[root@test01 ~]# curl -H "Content-Type:application/json" -X POST --data '{"username":"liuxin","password":"tiantiankaixin"}' http://192.168.1.30/api/useradd
true
这时候数据库中已经有了用户
文章图片
image.png
最后再把验证方式该回去
class UserAddApi(Resource):
# 添加用户，要求验证
@auth.login_required
2. 添加pxe信息
[root@test01 ~]# curl -H "Content-Type:application/json" -X POST --data '{"sn":"sn123456","pxe_ip":"10.64.115.i1","ilo_ip":"10.67.255.1","mac1":"aa:bb:cc:dd:dd:ee","mac2":"aa:bb:cc:dd:dd:e3","sw_name1":"sw_name1","sw_name2":"sw_name2","sw_port1":"sw_port1","sw_port2":"sw_port2"}' http://192.168.1.30/api/pxeinfo
{
"error": "Invalid credentials"
}
[root@test01 ~]# curl -H "Content-Type:application/json" -X POST --data '{"sn":"sn123456","pxe_ip":"10.64.115.i1","ilo_ip":"10.67.255.1","mac1":"aa:bb:cc:dd:dd:ee","mac2":"aa:bb:cc:dd:dd:e3","sw_name1":"sw_name1","sw_name2":"sw_name2","sw_port1":"sw_port1","sw_port2":"sw_port2"}' http://192.168.1.30/api/pxeinfo -u liuxin:tiantiankaixin
true
使用账号密码的认证方式，总是将账号密码在网络中传来传去，感觉不安全，怕被安全组同学截获，然后告诉我有漏洞，挨批评。所以我们使用token的方式验证
[root@test01 ~]# curl -H "Content-Type:application/json" -X GET http://192.168.1.30/api/token -u liuxin:tiantiankaixin
{
"error": "404 Not Found: The requested URL was not found on the server.If you entered the URL manually please check your spelling and try again."
}
[root@test01 ~]# curl -H "Content-Type:application/json" -X GET http://192.168.1.30/api/usertoken -u liuxin:tiantiankaixin
{
"token": "eyJhbGciOiJIUzI1NiIsImlhdCI6MTUyMjQ2MDAwNiwiZXhwIjoxNTIyNDYzNjA2fQ.eyJpZCI6MX0.hEh5_4OG3xuWzRiksG8w-E482korNMiO7yyHCFEkaHs"
}
[root@test01 ~]# curl -H "Content-Type:application/json" -X POST --data '{"sn":"sn123457","pxe_ip":"10.64.115.i1","ilo_ip":"10.67.255.1","mac1":"aa:bb:cc:dd:dd:ee","mac2":"aa:bb:cc:dd:dd:e3","sw_name1":"sw_name1","sw_name2":"sw_name2","sw_port1":"sw_port1","sw_port2":"sw_port2"}' http://192.168.1.30/api/pxeinfo -u eyJhbGciOiJIUzI1NiIsImlhdCI6MTUyMjQ2MDAwNiwiZXhwIjoxNTIyNDYzNjA2fQ.eyJpZCI6MX0.hEh5_4OG3xuWzRiksG8w-E482korNMiO7yyHCFEkaHs:
true
3. 查询pxe信息
[root@test01 ~]# curl -H "Content-Type:application/json" -X GET http://192.168.1.30/api/pxeinfo?sn=sn123456 -u eyJhbGciOiJIUzI1NiIsImlhdCI6MTUyMjQ2MDAwNiwiZXhwIjoxNTIyNDYzNjA2fQ.eyJpZCI6MX0.hEh5_4OG3xuWzRiksG8w-E482korNMiO7yyHCFEkaHs:
{
"id": 1,
"ilo_ip": "10.67.255.1",
"info_time": "Fri, 30 Mar 2018 21:28:16 GMT",
"mac1": "aa:bb:cc:dd:dd:ee",
"mac2": "aa:bb:cc:dd:dd:e3",
"pxe_ip": "10.64.115.i1",
"sn": "sn123456",
"sw_name1": "sw_name1",
"sw_name2": "sw_name2",
"sw_port1": "sw_port1",
"sw_port2": "sw_port2"
}
五、遗留的一些问题
1. token问题 用户可以使用旧token访问
http://192.168.1.30/api/usertoken申请新token。这也算是一个安全漏洞吧1. user问题 加密的token解密后定义死了是id=1的用户，所以id等于1的用户必须要有，而且所有使用token访问的用户都是同一个，不利于排查安全问题
推荐阅读
- PMSJ寻平面设计师之现代(Hyundai）
- 由浅入深理解AOP
- 【译】20个更有效地使用谷歌搜索的技巧
- 基于微信小程序带后端ssm接口小区物业管理平台设计
- mybatisplus如何在xml的连表查询中使用queryWrapper
- MybatisPlus|MybatisPlus LambdaQueryWrapper使用int默认值的坑及解决
- MybatisPlus使用queryWrapper如何实现复杂查询
- 爱琐搭配（喜欢复古、冷淡，像这种双环设计的气质耳环）
- iOS中的Block
- Linux下面如何查看tomcat已经使用多少线程