今天一起来通过 Flask 快速完成并部署一个 Restuful 服务，不要轻易走开哦！

一、框架概要

先来看看大致的代码框架。

这里说明下，这套代码结构是参照经典flask书籍《Flask Web Development》来写的，想要这本书电子版的，可以联系我哈。

二、代码详解

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.12，配置文件

在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_auth5，表模型

在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 "".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 的请求消息体，并且对应的更新数据库中表的值。

三、初始化

使用在 manage 中定义的 shell 命令，进入 shell 界面。

执行 db.create_all() 创建在 models 中定义的表，执行成功之后，查看数据库如下：

表已经创建成功，但是我们的 admin 用户还不存在，继续执行命令 AdminUser.init_user()，插入 Admin 用户。

到此，初始化工作完毕。

四、测试API

执行命令 python manage.py 启动 flask 服务，可以看到已经在本地的9980端口启动。

使用 postman 工具测试 useraddAPI，选择 POST 方式，鉴权方式使用 Basic Auth，body 中使用 json 形式。

发送请求测试如下：

如果修改名字为错误值，都会鉴权失败。