Garland +

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

背景

厂里的几个后台大部分是管理员和商家共用的,而且接口的命名都很混乱,对于资源的增删改查,大部分都是放在查询字符串或者 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 真好使。

Blog

Thoughts

Project