今天一起来通过 Flask 快速完成并部署一个 Restuful 服务，不要轻易走开哦
01.框架概要
先来看看大致的代码框架
这里说明下，这套代码结构是参照经典flask书籍《Flask Web Development》来写的，想要这本书电子版的，可以联系我哈。
02.代码详解
1，依赖包
放到了requirements文件里
flask==1.0.2 flask-script==2.0.6 flask-restful==0.3.7 flask-sqlalchemy==2.3.3 flask-httpauth==3.2.4 itsdangerous==1.1.0 Flask-SQLAlchemy==2.3.2 psycopg2==2.7.6.1
2，配置文件
在config.py文件里
import os basedir = os.path.abspath(os.path.dirname(__file__)) class Config: @staticmethod def init_app(app): pass class DevelopmentConfig(Config): pass class TestingConfig(Config): TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'my.sqlite3') class ProductionConfig(Config): SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') config = { 'development': DevelopmentConfig, 'testing': TestingConfig, 'production': ProductionConfig, 'default': DevelopmentConfig }
这里我使用了简单易用的 sqlite 数据库，当然也可以使用其他关系型数据库，比如 MySQL 等。
3，程序入口文件
manage.py
from app import create_app, db from flask_script import Manager, Shell, Server from app.models import User, AdminUser, Picture app = create_app('testing') manager = Manager(app) def make_shell_context(): return dict(app=app, db=db, User=User, AdminUser=AdminUser, Picture=Picture) manager.add_command("runserver", Server(use_debugger=True, host='0.0.0.0', port='9980')) manager.add_command("shell", Shell(make_context=make_shell_context)) if __name__ == '__main__': manager.run(default_command='runserver')
引入了 flask 的 command 用法，主要用来做数据库的初始化操作和快速启动 flask 服务。
4，引入蓝图
蓝图可以帮我们实现模块化应用的功能。
在主程序的__init__.py文件中创建 flask app 并注册模块
from flask import Flask from flask_sqlalchemy import SQLAlchemy from config import config db = SQLAlchemy() def create_app(config_name): app = Flask(__name__) app.config.from_object(config[config_name]) config[config_name].init_app(app) db.init_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
同时在 api 模块的__init__.py里引入蓝图
from flask import Blueprint api_1_0 = Blueprint('api_1_0', __name__, url_prefix='/api') from . import api_user, api_auth
5，表模型
在models.py文件里，定义了当前用到的数据库表结构
from . import db from werkzeug.security import generate_password_hash, check_password_hash class AdminUser(db.Model): __tablename__ = 'adminuser' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(128), unique=True, index=True) password_hash = db.Column(db.String(128)) @staticmethod def init_user(): users = AdminUser.query.filter_by(username='admin').first() if users is None: users = AdminUser(username='admin') users.password = 'hardtoguess' db.session.add(users) db.session.commit() @property def password(self): raise AttributeError('password is not readable attribute') @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 "<Users {}>".format(self.username) class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(128), unique=True, index=True) picture_count = db.Column(db.Integer) class Picture(db.Model): __tablename__ = 'pictures' id = db.Column(db.Integer, primary_key=True) picture_id = db.Column(db.Integer, db.ForeignKey('users.id')) picture_name = db.Column(db.String(128)) picture = db.Column(db.Text)
其中 AdminUser 是作为 API 的鉴权用户的存在，后面调用 API 都会使用该用户，而 User 和 Picture 则是后面需要操作的表了。
着重说下 AdminUser 类，定义了一个静态方法 init_user，是用来后面初始化数据库的，我们需要把这个鉴权用户手动添加到数据库中。方法 verify_password 用于后面 API 的鉴权，如果数据库存在该用户且密码的哈希值相同，则鉴权通过。
6，API部分
api 鉴权代码，使用 flask_httpauth 库
from flask_httpauth import HTTPBasicAuth from flask import jsonify from ..models import AdminUser auth = HTTPBasicAuth() @auth.error_handler def unauthorized(): error_info = '{}'.format("Invalid credentials") print(error_info) response = jsonify({'error': error_info}) response.status_code = 403 return response @auth.verify_password def verify_password(username, password): user = AdminUser.query.filter_by(username=username).first() if not user or not user.verify_password(password): return False return True
先到数据库查询用户，如果不存在或者密码哈希值不正确，则返回 False，鉴权失败。
7，设计API
在api_user.py文件中，初始化 flask_restful 的 Api 类，用于后面添加 resource
api_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']) p = Picture(picture_name=user_info['username']) u.password = user_info['password'] u.picture_count = user_info['picture_count'] db.session.add(u) db.session.add(p) db.session.commit() except: logger.info("User add: {} failure...".format(user_info['username'])) db.session.rollback() return False else: logger.info("User add: {} success...".format(user_info['username'])) return True finally: db.session.close() api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')
这里只接受 json 的请求消息体，并且对应的更新数据库中表的值。
03.初始化
使用在 manage 中定义的 shell 命令，进入 shell 界面
执行 db.create_all() 创建在 models 中定义的表，执行成功之后，查看数据库如下
表已经创建成功，但是我们的 admin 用户还不存在，继续执行命令 AdminUser.init_user()，插入 Admin 用户。
到此，初始化工作完毕
04.测试API
执行命令 python manage.py 启动 flask 服务，可以看到已经在本地的9980端口启动
使用 postman 工具测试 useraddAPI，选择 POST 方式，鉴权方式使用 Basic Auth，body 中使用 json 形式
发送请求测试如下：
如果修改名字或者密码为错误值，都会鉴权失败
至此，我们的一个简易 Restful 服务就搭建完成了，喜欢就点个赞吧~
