34. Flask 集成 flask-restful

需求

一般 Flask 框架默认写 api 的请求就是写一个函数视图，如下：

@app.route('/add_list', methods=["POST"])
def add_list():
	...
    
@app.route('/get_list', methods=["GET"])
def get_list():
    ...

这种方式当然可以开发 api， 但是当我们想要基于 restful 风格来编写 api，就不太方便了。就需要写 4 个单独的函数视图，如下：

# result api
# 路由 /list
# 增 POST
# 删 DELETE
# 查 GET
# 改 PUT
 
# 具体函数方法如下：就要写4个分开的函数，这样就感觉一种不太好的感觉。
@app.route('/list', methods=["POST"])
def add_list():
	...
    
@app.route('/list', methods=["GET"])
def get_list():
    ...
 
@app.route('/list', methods=["DELETE"])
def delete_list():
    ...
 
@app.route('/list', methods=["PUT"])
def save_list():
    ...

有没有一种更加简洁的方法呢？ 例如：

class List(restful.Resource):
	
	# get 查询
    def get(self):
    
    # post 新增
    def post(self):
        
    # delete 删除
    def delete(self):     
    
    # put 修改
    def put(self):    

其实也就是跟 Django 的类视图一样的开发方式了，下面我们就要介绍一下使用 flask-resutful 库来完成这个需求。

参考文档

http://www.pythondoc.com/flask-restful/first.html

https://flask-restful.readthedocs.io/en/latest/

安装 Flask-Restful

使用 pip 安装：

$ pip install flask-restful

快速入门示例

1. 一个很小的 Flask-RESTful API 示例

from flask import Flask
from flask_restful import Resource, Api # 导入flask_resutful
 
app = Flask(__name__)
api = Api(app) # 创建 api
 
# 编写api的请求业务: api资源
class HelloWorld(Resource):
    
    # get请求
    def get(self):
        return {'hello': 'world'} # 简化了json格式返回，等价于 return jsonify({'hello': 'world'})
 
# 设置api的url路径：api 路径
api.add_resource(HelloWorld, '/')
 
if __name__ == '__main__':
    app.run(debug=True)

启动服务:

$ python api.py
 * Running on http://127.0.0.1:5000/
 * Restarting with reloader

测试如下：

2. 配置资源的路由 Resourceful Routing

上面我们已经写了一个最简单的 flask-restful api 示例，下面来增加多 put 请求，并且统一可以配置 资源的路由 Resourceful Routing， 示例如下：

from flask import Flask, request
from flask_restful import Resource, Api # 导入flask_resutful
 
app = Flask(__name__)
api = Api(app) # 创建 api
 
# 创建todos
todos = {}
 
# 编写Api资源
class TodoSimple(Resource):
 
    # get请求
    def get(self, todo_id):
        return {todo_id: todos[todo_id]}
 
    # put请求
    def put(self, todo_id):
        todos[todo_id] = request.form['data']
        return {todo_id: todos[todo_id]}
 
# 配置资源的路由
api.add_resource(TodoSimple, '/<string:todo_id>')
 
if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port="5000")

可以看到，我们写了两个请求 get 和 put，两个请求都是同一个 url 路径，下面使用 curl 测试如下：

# 执行put请求，设置 todo1
[root@dev ~]# curl http://10.120.10.241:5000/todo1 -d "data=Remember the milk" -X PUT
{
    "todo1": "Remember the milk"
}
[root@dev ~]# 
# 执行get请求，查询 todo1
[root@dev ~]# curl http://10.120.10.241:5000/todo1 -X GET
{
    "todo1": "Remember the milk"
}
[root@dev ~]# 
# 执行put请求，设置 todo2 
[root@dev ~]# curl http://10.120.10.241:5000/todo2 -d "data=Change my brakepads" -X PUT
{
    "todo2": "Change my brakepads"
}
[root@dev ~]# 
# 执行get请求，查询 todo2
[root@dev ~]# curl http://10.120.10.241:5000/todo2 -X GET
{
    "todo2": "Change my brakepads"
}
[root@dev ~]# 

3.设置返回响应的 响应体、 响应码 以及 响应头

与 Flask 的返回响应一致， Flask Restful 设置的返回也是按照如下格式设置响应的：

return 响应体, 状态码, 响应头

下面只要再写一个API即可示例：

# 设置响应信息示例
class ReponseInfo(Resource):
    def get(self):
        # 响应体, 状态码, 响应头
        return {'task': 'Hello world'}, 201, {'Etag': 'some-opaque-string', 'city': 'shenzhen'}
 
api.add_resource(ReponseInfo, '/res')

启动服务之后，使用 curl 测试如下：

[root@dev ~]# curl -i http://10.120.10.241:5000/res -X GET
HTTP/1.0 201 CREATED # http响应码 201
Content-Type: application/json
Content-Length: 30
Etag: some-opaque-string # 自定义的响应 header
city: shenzhen           # 自定义的响应 header
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Fri, 18 Sep 2020 09:23:04 GMT
# 返回的响应体
{
    "task": "Hello world"
}
[root@dev ~]# 

4. 设置API的路径，也就是资源的端点 Endpoints

4.1 配置多个URL至同一个Api资源

有些使用对于一个Api资源可能会有多个 url 路径进行访问，例如：访问首页可能使用 / 或者 /index 都可以访问，那么这个 Api 的端点可以配置示例如下：

# 设置首页
class Index(Resource):
 
    def get(self):
        return {'msg': 'index'}
 
# 配置多个url路径到访问首页
api.add_resource(Index, '/', '/index')

启动服务之后，使用 curl 测试两个路径如下：

[root@dev ~]# curl -i http://10.120.10.241:5000/ -X GET
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 23
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Fri, 18 Sep 2020 09:36:02 GMT
 
{
    "msg": "index"
}
[root@dev ~]# 
[root@dev ~]# curl -i http://10.120.10.241:5000/index -X GET
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 23
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Fri, 18 Sep 2020 09:36:06 GMT
 
{
    "msg": "index"
}
[root@dev ~]# 

可以看到两个URL的路径都能够正常访问 index

4.2 设置 url 的命名端点 endpoint

跟Django的命名路由 url 一样，我们也可以使用参数给 endpoint 进行命名，然后使用 flask-restful 库中的 url_for() 来解析 url 路径，示例代码如下：

from flask_restful import url_for # 导入flask_resutful的url，注意不是 flask 的 url_for
 
# 设置首页
class Index(Resource):
 
    def get(self):
        return {
            'msg': 'index',
            'url': url_for('index') # 使用url_for解析出命名为 index 的路径，如果有多个路径，则返回第一个
        }
 
# 配置多个url路径到访问首页
api.add_resource(Index, '/', '/index', endpoint='index')

启动服务，使用 curl 测试如下：

[root@dev ~]# curl -i http://10.120.10.241:5000/index -X GET 
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 39
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Fri, 18 Sep 2020 10:57:54 GMT
 
{
    "msg": "index",
    "url": "/"       # 就算请求的是 /index 路径，但是 url_for 返回的还是第一个 / 路径
}
[root@dev ~]# 

5. 设置认证修饰器

之前我们使用函数写视图方法的时候，是比较方便写一些修饰器的，那么如果我们需要给 类视图资源 设置修饰器，该怎么办呢？

下面我们使用 HTTP 的 BasicAuth 来写一个登陆的修饰器使用。

5.1. 安装BasicAuth需要的库

pip install Flask-HTTPAuth

5.2 导入BasicAuth的相关库

from flask_httpauth import HTTPBasicAuth
from werkzeug.security import generate_password_hash, check_password_hash

5.3 视图函数使用认证如下

from flask import Flask, request, make_response, jsonify
from flask_restful import Resource, Api, url_for # 导入flask_resutful
from flask_httpauth import HTTPBasicAuth
from werkzeug.security import generate_password_hash, check_password_hash
 
 
app = Flask(__name__)
api = Api(app) # 创建 api
auth = HTTPBasicAuth() # 创建BasicAuth对象
 
# 用户的名称以及密码
users = {
    "john": generate_password_hash("hello"),
    "susan": generate_password_hash("bye")
}
 
# 验证用户密码的修饰器
@auth.verify_password
def verify_password(username, password):
    if username in users and check_password_hash(users.get(username), password):
        return username
 
# 自定义未认证通过的返回
@auth.error_handler
def unauthorized():
    return make_response(jsonify({'error': 'Unauthorized access'}), 401)
 
 
# 设置首页
class Index(Resource):
 
    # 设置登陆的修饰器
    decorators = [auth.login_required]
 
    def get(self):
        return {
            'msg': 'index',
            'url': url_for('index') # 使用url_for解析出命名为 index 的路径，如果有多个路径，则返回第一个
        }
 
# 配置多个url路径到访问首页
api.add_resource(Index, '/', '/index', endpoint='index')
 
if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port="5000")

启动服务之后，下面我们使用 curl 来测试一下：

# 没有设置用户名、密码，则返回认证失败
[root@dev ~]# curl  -i http://10.120.10.241:5000/
HTTP/1.0 401 UNAUTHORIZED
Content-Type: application/json
Content-Length: 37
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Mon, 21 Sep 2020 03:03:09 GMT
 
{
  "error": "Unauthorized access"
}
[root@dev ~]# 
 
# 设置用户名、密码，成功访问
[root@dev ~]# curl -u john:hello  -i http://10.120.10.241:5000/
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 39
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Mon, 21 Sep 2020 03:04:02 GMT
 
{
    "msg": "index",
    "url": "/"
}
[root@dev ~]# 

6.设置多个自定义修饰器

上面我们已经成功使用上的 BasicAuth修饰器， 我们再自定义一个简单的修饰器，添加到 资源视图类 中。

6.1 编写一个自定义修饰器

import functools
 
# 自定义的装饰器
def test_required(view_func):
    # wraps函数的作用是将wrapper内层函数的属性设置为被装饰函数view_func的属性
    @functools.wraps(view_func)
    def wrapper(*args, **kwargs):
 
        print("=====执行自定义装饰器==========")
        return view_func(*args, **kwargs)
 
    return wrapper

6.2 在资源视图类添加使用修饰器 test_required

from flask import Flask, request, make_response, jsonify
from flask_restful import Resource, Api, url_for # 导入flask_resutful
from flask_httpauth import HTTPBasicAuth
from werkzeug.security import generate_password_hash, check_password_hash
import functools
 
app = Flask(__name__)
api = Api(app) # 创建 api
auth = HTTPBasicAuth() # 创建BasicAuth对象
 
# 用户的名称以及密码
users = {
    "john": generate_password_hash("hello"),
    "susan": generate_password_hash("bye")
}
 
# 验证用户密码的修饰器
@auth.verify_password
def verify_password(username, password):
    if username in users and check_password_hash(users.get(username), password):
        return username
 
# 自定义未认证通过的返回
@auth.error_handler
def unauthorized():
    return make_response(jsonify({'error': 'Unauthorized access'}), 401)
 
# 自定义的装饰器
def test_required(view_func):
    # wraps函数的作用是将wrapper内层函数的属性设置为被装饰函数view_func的属性
    @functools.wraps(view_func)
    def wrapper(*args, **kwargs):
 
        print("=====执行自定义装饰器==========")
        return view_func(*args, **kwargs)
 
    return wrapper
 
# 设置首页
class Index(Resource):
 
    # 设置登陆的修饰器\自定义修饰器
    decorators = [auth.login_required, test_required]
 
    def get(self):
        return {
            'msg': 'index',
            'url': url_for('index') # 使用url_for解析出命名为 index 的路径，如果有多个路径，则返回第一个
        }
 
# 配置多个url路径到访问首页
api.add_resource(Index, '/', '/index', endpoint='index')
 
 
if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port="5000")

启动服务之后，使用 curl 测试如下：

[root@dev ~]# curl -u john:hello  -i http://10.120.10.241:5000/
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 39
Server: Werkzeug/0.16.0 Python/3.7.2
Date: Mon, 21 Sep 2020 03:17:21 GMT
 
{
    "msg": "index",
    "url": "/"
}
[root@dev ~]# 

同时后台打印 自定义修饰器的 信息：

验证集成 Flask-Restful 处理 GET POST 的请求参数

上面我们已经基本了解集成 Flask-Restful 的使用了，那么集成了之后，对于GET请求的query参数获取、POST请求的表单或者json参数获取，有什么地方要注意的么？下面我们来写一个演示实例。

1.示例代码

from flask import Flask, request
from flask_restful import Resource, Api, url_for  # 导入flask_resutful
import json
 
# 设置集成Flask-Resutful的POST请求
class Requests(Resource):
 
    def get(self):
        """接收处理query参数： ?user_name=libai&user_age=17"""
 
        # 接收处理GET数据请求
        user_name = request.args.get('user_name')
        user_age = request.args.get('user_age')
 
        return {"user_name": user_name, "user_age": user_age}
 
 
    def post(self):
        """接收处理json数据请求"""
 
        data = json.loads(request.data)  # 将json字符串转为dict
        user_name = data['user_name']
        user_age = data['user_age']
 
        return {"user_name": user_name, "user_age": user_age}
 
 
api.add_resource(Requests, '/requests', endpoint='request')
 
if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port="5000")

2.使用 postman 测试 GET 请求，获取 query 参数的情况

可以看到正常获取参数。

3.使用 postman 测试 POST 请求，获取 json 请求体参数的情况

也是能够正常获取参数。

4.总结：

获取 query 参数 或者 json请求体参数，都是从 flask 库的 request 中获取，集成 Flask-Restful 并不影响使用。

from flask import request

验证集成 Flask-Restful 以及 蓝图 BluePrint

使用了 Flask-Restful 后，定义路由的方式就不同了一些，那么会不会影响蓝图的使用呢？ 下面来写个示例看看。

注意：在蓝图中，如果使用Flask_RESTful，那么在创建Api对象的时候，使用蓝图对象，不再是使用app对象了.

1.创建一个 admin 的蓝图应用

from flask_restful import Resource, Api  # 导入flask_resutful
from flask import Blueprint
 
#Blueprint必须指定两个参数，admin表示蓝图的名称，__name__表示蓝图所在模块
admin = Blueprint('admin',__name__)
 
api = Api(admin) # 注意： 使用蓝图对象来创建 api
 
# API业务
class AdminView(Resource):
 
    def get(self):
        return {"msg": "admin index"}
 
# 配置url路径
api.add_resource(AdminView, '/', endpoint='admin')

可以从代码看到，本来 flask_restful 的 Api 创建是需要 flask 的 app 的，这里就采用 蓝图对象 而已，其他使用上没有什么区别。

2.创建 app，注册蓝图应用

from flask import Flask
 
# 创建app
app = Flask(__name__)
 
# 注册蓝图
from .admin import admin
app.register_blueprint(admin,url_prefix='/admin')
 
if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port="5000")

3. 使用postman测试如下