写一个 flask 扩展路由功能的小插件

2017-08-17

背景

厂里的几个后台大部分是管理员和商家共用的，而且接口的命名都很混乱，对于资源的增删改查，大部分都是放在查询字符串或者 post 表单里，而且命名也不统一，所以对于同一个资源的每个接口，就会写大量的参数校验，权限校验，不多说，举个例子。

管理员 /manage/service/admin/api/finish_live.json
主播 /manage/service/shops/api/finish_live.json

上面是我们后台老的 API 结构，层级基本可以分为 /后台/业务/身份/uri类型/具体行为，对于同一个资源会有很多行为，就拿这个 live 来讲

获取直播绑定的业务: /live_obj_show.json
获取推流地址: /live_push_url.json
结束直播: /live_finish.json
获取直播弹幕: /live_danmaku.json
在这个直播禁言用户: /live_forbid_send_msg.json
获取这个直播被禁言的用户: /live_group_shutted_user.json
在这个直播解禁用户: /live_unforbid_send_msg.json

详细讲的话还有更多，这时遇到了第一个问题，权限校验，管理员有不同的权限，主播也有不同的权限，有的权限是只读，有的权限是可读可写，这样就写了大量的装饰器，然后每个接口都要加上权限装饰器，这样写很恶心，装饰器太多的话，后面新来的人根本没办法维护，而且最主要的是，新加的接口很容易忘了或者用错了装饰器。

第二个问题是前面说的，资源ID 都是放在查询字符串或者 post 表单里，所以每个接口首先得判断有没有这个资源，然后再判断当前登录的用户有没有对这个资源操作的权限，这样就会写大量的差不多但是又有那么一点点不一样的代码。

解决办法

定义好接口命名

最后定义的接口大概是这样的格式 /后台/业务/身份/uri类型/资源/id/行为，具体到一开始的例子就是:

管理员 /manage/service/admin/api/live/<int:id>/finish.json
主播 /manage/service/shops/api/live/<int:id>/finish.json

把校验部分抽离

利用 url 的树形结构我们可以画这样一张图

        /manage(+validator)
         /\
        /  \
 /service1 /service2
              /\
             /  \
      /shops/api /admin/api(+validator)
                    /\
                   /  \           |route('/push_url.json')
        /course/<id> /live/<id> --|route('/finish.json')
                                  |route('/danmaku.json')

权限校验可以在各个节点上加校验器，在拼装 path 的时候就做校验，没有权限的直接返回；对于资源的 校验在资源节点就判断资源的有无，当前用户对资源有没有权限，校验失败就直接返回，这样对于具体的 view 函数内就只用处理内部的逻辑。

实现方式

前前后后大概实现了三种方法，最后用了最后一种

构造一个全局的校验树

大体思路是，在 flask app 初始化的时候构造出一颗 path 树，每个节点的名称是 path 对应的节点名，上面有各自的校验器。新的请求在 app.before_request 的时候会从上到下一级一级进行校验，失败了就直接返回。

当时是这样的实现:

# coding: utf-8
# /privilege.py
from flask import g
class _Node(object):
    def __init__(self, name):
        self.name = name
        self.validator = None
        self.children = {}
class Validators(object):
    @classmethod
    def privilege_a_required(cls):
        if not g.me:
            return False
        if not g.me.has_privilege('a'):
            return False
        return True
    @classmethod
    def privilege_b_required(cls):
        if not g.me:
            return False
        if not g.me.has_privilege('b'):
            return False
        return True
    @classmethod
    def privilege_c_required(cls):
        if not g.me:
            return False
        if not g.me.has_privilege('c'):
            return False
        return True
_paths = {
    Validators.privilege_a_required: [
        '/manage/service1/god'
    ],
    Validators.privilege_b_required: [
        '/manage/service1/god/api/courses/create.json',
        '/manage/service1/god/api/courses/edit.json',
    ],
    Validators.privilege_c_required: [
        '/manage/service1/god/api/courses/show.json',
        '/manage/service1/god/api/courses/search.json'
    ],
}
def build_tree_from_paths():
    root = _Node('manage')
    for validator, paths in _paths.items():
        for path in paths:
            path_dirs = path.lstrip('/').rstrip('/').split('/')
            father = root
            for idx, dir_name in enumerate(path_dirs):
                if dir_name not in father.children:
                    father.children[dir_name] = _Node(dir_name)
                father = father.children[dir_name]
                if idx == len(path_dirs) - 1:
                    father.validator = validator
    return root
def validate_path(path):
    path_dirs = path.lstrip('/').rstrip('/').split('/')
    if path_dirs[0] != g.tree.name:
        return True
    father = g.tree
    for dir_name in path_dirs:
        if dir_name not in father.children:
            return True
        father = father.children[dir_name]
        validator = father.validator
        if validator is None:
            continue
        if not validator():
            return False
    return True

当时在思考这种实现的时候还沿用的是之前的 path 结构。首先定义树的节点，节点名是 path 节点去掉 / 的名称，每个节点上可以绑定一个校验器，校验器的返回值是 True 或者 False，还有它的儿子们。

定义一堆校验器，将校验器和对应使用该校验器的 path 列出来构造成一个 map。这个 map 就是 flask app 初始化的时候构造出的树，校验器可以加在 path 任意一个节点上。使用方式如下

# coding: utf-8
import json
from flask import Flask, request, url_for, abort
from privilege import build_tree_from_paths, validate_path
app = Flask(__name__)
app.tree = build_tree_from_paths()
@app.before_request
def before_request():
    # load user
    g.tree = app.tree
    if not validate_path(request.path):
        if request.path.endswith('json'):
            return json.dumps({'status': 'error'})
        return abort(401)
if __name__ == '__main__':
    app.run(debug=True, port=9000)

在 app 初始化的时候 load 一下树，每个请求来的时候在最后回去根据 path 一级一级校验，这样其实是可以丢掉大部分的权限装饰器的。

不过这样还是有一些问题，一个是需要维护一个校验器的 map ，每次新加 view 函数都要去加一条， 这个很容易忘记，忘记了也不好发现；另一个是没有解决上面说的第二个问题。

Nestable blueprints

既然手动构造的话会忘记，所以就有了第二种方法，用 blueprint，blueprint 的好处是很多的， 可以 route 可以 before/after_request，如果它支持 blueprint of blueprint 的话。

你能想到的，别人一定早想过了！

我找了下 Falsk 的 issue，果然发现了一个: Nestable blueprints ,继承了下 Flask 的 Blueprint 加上了自定义的 register_blueprint 方法，可以让 blueprint register blueprint，实现如下：

# coding: utf-8
from flask import Blueprint
class CustomBlueprint(Blueprint):
    def register_blueprint(self, blueprint, **options):
        def deferred(state):
            url_prefix = (state.url_prefix or u"") + (options.get('url_prefix', blueprint.url_prefix) or u"")
            if 'url_prefix' in options:
                del options['url_prefix']
            state.app.register_blueprint(blueprint, url_prefix=url_prefix, **options)
        self.record(deferred)

然后可以把定义都放到一起，像下面这样

# coding: utf-8
# route.py
import json
from flask import request, abort, g
from common.utils.render import error
from libs.blueprint import CustomBlueprint
base_bp = CustomBlueprint('manage', __name__)
service1_bp = CustomBlueprint('service1', __name__)
god_api_bp = CustomBlueprint('god_api_bp', __name__)
courses_bp = CustomBlueprint('courses', __name__)
base_bp.register_blueprint(service1_bp, url_prefix='/service1')
service1_bp.register_blueprint(god_api_bp, url_prefix='/god/api')
god_api_bp.register_blueprint(courses_bp, url_prefix='/courses')
@courses_bp.before_request
def privilege():
    if not g.me or not g.me.has_privilege('a'):
        if request.path.endswith('json'):
            return json.dumps({'status': 'error'})
        return abort(401)

使用的时候

# coding: utf-8
import json
from flask import Flask, request, url_for, abort
from route import base_bp, courses_bp
app = Flask(__name__)
app.register_blueprint(base_bp, url_prefix='/manage')
@app.before_request
def before_request():
    # load user
@courses_bp.route('/show.json')
def service1_god_api_courses_all():
    pass
if __name__ == '__main__':
    app.run(debug=True, port=9000)

这个方法的好处是，开发新增的资源接口时不用去考虑上层的权限校验，也不会出现忘记的情况，因为不再使用 app.route() 了， 不过老大嫌这种方式会改变对原本的 blueprint 的理解，希望不依赖 flask 的东西做成一个纯插件的形式，所以这个方法也就此作罢

纯插件形式的 route 实现

不能用 blueprint 那就只能自己实现以下这个 route 方法了。思路还是从上向下可以一级一级加节点，校验部分参考了 before/after_request 的实现，自己实现的 route 最后还是要传给 flask 的 route 所以需要把 flask 的 app 对象传进去，代码如下：

# coding: utf-8
from functools import wraps
def setupmethod(f):
    def wrapper_func(self, *args, **kwargs):
        return f(self, *args, **kwargs)
    return wrapper_func
class Kotori(object):
    def __init__(self, app, prefix=None, father=None):
        """
        :param: app: flask application object
        :param: prefix: path prefix
        :param: father: father object of current node
        """
        self.app = app
        self.prefix = prefix or ''
        self.father = father
        self._path = None
        self.before_request_funcs = []
        self.after_request_funcs = []
    def add_prefix(self, prefix):
        """add child path prefix
        :param: prefix: node of url path
        """
        return Kotori(self.app, prefix, self)
    @property
    def path(self):
        """get path of current node"""
        if self._path is None:
            x = [self.prefix]
            k = self
            while k.father is not None:
                x.append(k.father.prefix)
                k = k.father
            self._path = ''.join(reversed(x))
        return self._path
    @setupmethod
    def before_request(self, f):
        """register a function for current node to run before each request"""
        self.before_request_funcs.append(f)
    @setupmethod
    def after_request(self, f):
        """register a function for current node to run after each request"""
        self.after_request_funcs.append(f)
    def _run_route_func(self, func_attribute_name):
        funcs = []
        current = self
        while current.father is not None:
            for f in getattr(current, func_attribute_name, []):
                funcs.append(f)
            current = current.father
        for f in reversed(funcs):
            res = f()
            if res is not None:
                return res
    def _run_after_request(self):
        return self._run_route_func('after_request_funcs')
    def _run_before_request(self):
        return self._run_route_func('before_request_funcs')
    def route(self, rule, **options):
        """a decorator that is used to register a view function for a given end-url-rule,
        params are the same as Flask.route
        """
        rule = self.path + rule
        def _(f):
            @self.app.route(rule, **options)
            @wraps(f)
            def deco(*args, **kwargs):
                before_request_res = self._run_before_request()
                if before_request_res is not None:
                    return before_request_res
                res = f(*args, **kwargs)
                after_request_res = self._run_after_request()
                if after_request_res is not None:
                    return after_request_res
                return res
            return deco
        return _

使用方式如下：

# coding: utf-8
import json
from flask import Flask
from kotori import Kotori
app = Flask(__name__)
kotori_app = Kotori(app)
manage_app = kotori_app.add_prefix('/manage')
@app.before_request
def before_request():
    # load user
service1 = manage_app.add_prefix('/service1')
god_api = service1.add_prefix('/god/api')
courses = god_api.add_prefix('/courses/<int:id>')
@service1.before_request
def service1_validator():
    # do some validation
    pass
@courses.before_request
def courses_validator():
    # do some validation
    pass
@courses.route('/show.json')
def service1_god_api_courses_all():
    pass
if __name__ == '__main__':
    app.run(debug=True, port=9000)

完全的纯插件形式，从上向下一级一级加节点，可以在每个节点加入多个 hook，完全兼容 flask 的 route 格式， 资源ID放在 url 里可以在具体操作的父节点就进行校验，避免了判断资源是否存在、当前用户对资源操作权限的代码 在每个 view 函数都写一遍，后期也可以对参数校验的部分增加更详细的控制，而且对于 url path 的格式也能规范。

这个模块先是在我们的课程直播后台试了下，也把之前的老代码慢慢迁移过来，去掉了之前的不少痛点，hhh 真好使。