Garland +

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

2017-08-17

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

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

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

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

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

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

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

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

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

  1.         /manage(+validator)
  2.          /\
  3.         /  \
  4.  /service1 /service2
  5.               /\
  6.              /  \
  7.       /shops/api /admin/api(+validator)
  8.                     /\
  9.                    /  \           |route('/push_url.json')
  10.         /course/<id> /live/<id> --|route('/finish.json')
  11.                                   |route('/danmaku.json')

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

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

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

当时是这样的实现:

  1. # coding: utf-8
  2. # /privilege.py
  4. from flask import g
  7. class _Node(object):
  9.     def __init__(self, name):
  10.         self.name = name
  11.         self.validator = None
  12.         self.children = {}
  15. class Validators(object):
  17.     @classmethod
  18.     def privilege_a_required(cls):
  19.         if not g.me:
  20.             return False
  21.         if not g.me.has_privilege('a'):
  22.             return False
  23.         return True
  25.     @classmethod
  26.     def privilege_b_required(cls):
  27.         if not g.me:
  28.             return False
  29.         if not g.me.has_privilege('b'):
  30.             return False
  31.         return True
  33.     @classmethod
  34.     def privilege_c_required(cls):
  35.         if not g.me:
  36.             return False
  37.         if not g.me.has_privilege('c'):
  38.             return False
  39.         return True
  42. _paths = {
  43.     Validators.privilege_a_required: [
  44.         '/manage/service1/god'
  45.     ],
  46.     Validators.privilege_b_required: [
  47.         '/manage/service1/god/api/courses/create.json',
  48.         '/manage/service1/god/api/courses/edit.json',
  49.     ],
  50.     Validators.privilege_c_required: [
  51.         '/manage/service1/god/api/courses/show.json',
  52.         '/manage/service1/god/api/courses/search.json'
  53.     ],
  54. }
  57. def build_tree_from_paths():
  58.     root = _Node('manage')
  59.     for validator, paths in _paths.items():
  60.         for path in paths:
  61.             path_dirs = path.lstrip('/').rstrip('/').split('/')
  62.             father = root
  63.             for idx, dir_name in enumerate(path_dirs):
  64.                 if dir_name not in father.children:
  65.                     father.children[dir_name] = _Node(dir_name)
  66.                 father = father.children[dir_name]
  67.                 if idx == len(path_dirs) - 1:
  68.                     father.validator = validator
  69.     return root
  72. def validate_path(path):
  73.     path_dirs = path.lstrip('/').rstrip('/').split('/')
  74.     if path_dirs[0] != g.tree.name:
  75.         return True
  77.     father = g.tree
  78.     for dir_name in path_dirs:
  79.         if dir_name not in father.children:
  80.             return True
  81.         father = father.children[dir_name]
  82.         validator = father.validator
  83.         if validator is None:
  84.             continue
  85.         if not validator():
  86.             return False
  87.     return True

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

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

  1. # coding: utf-8
  3. import json
  4. from flask import Flask, request, url_for, abort
  5. from privilege import build_tree_from_paths, validate_path
  7. app = Flask(__name__)
  8. app.tree = build_tree_from_paths()
  11. @app.before_request
  12. def before_request():
  13.     # load user
  14.     g.tree = app.tree
  15.     if not validate_path(request.path):
  16.         if request.path.endswith('json'):
  17.             return json.dumps({'status': 'error'})
  18.         return abort(401)
  21. if __name__ == '__main__':
  22.     app.run(debug=True, port=9000)

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

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

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

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

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

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

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

  1. # coding: utf-8
  2. # route.py
  3. import json
  4. from flask import request, abort, g
  6. from common.utils.render import error
  8. from libs.blueprint import CustomBlueprint
  10. base_bp = CustomBlueprint('manage', __name__)
  11. service1_bp = CustomBlueprint('service1', __name__)
  12. god_api_bp = CustomBlueprint('god_api_bp', __name__)
  13. courses_bp = CustomBlueprint('courses', __name__)
  15. base_bp.register_blueprint(service1_bp, url_prefix='/service1')
  16. service1_bp.register_blueprint(god_api_bp, url_prefix='/god/api')
  17. god_api_bp.register_blueprint(courses_bp, url_prefix='/courses')
  20. @courses_bp.before_request
  21. def privilege():
  22.     if not g.me or not g.me.has_privilege('a'):
  23.         if request.path.endswith('json'):
  24.             return json.dumps({'status': 'error'})
  25.         return abort(401)

使用的时候

  1. # coding: utf-8
  3. import json
  4. from flask import Flask, request, url_for, abort
  5. from route import base_bp, courses_bp
  7. app = Flask(__name__)
  8. app.register_blueprint(base_bp, url_prefix='/manage')
  10. @app.before_request
  11. def before_request():
  12.     # load user
  15. @courses_bp.route('/show.json')
  16. def service1_god_api_courses_all():
  17.     pass
  20. if __name__ == '__main__':
  21.     app.run(debug=True, port=9000)

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

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

  1. # coding: utf-8
  3. from functools import wraps
  6. def setupmethod(f):
  7.     def wrapper_func(self, *args, **kwargs):
  8.         return f(self, *args, **kwargs)
  9.     return wrapper_func
  12. class Kotori(object):
  14.     def __init__(self, app, prefix=None, father=None):
  15.         """
  16.         :param: app: flask application object
  17.         :param: prefix: path prefix
  18.         :param: father: father object of current node
  19.         """
  20.         self.app = app
  21.         self.prefix = prefix or ''
  22.         self.father = father
  23.         self._path = None
  24.         self.before_request_funcs = []
  25.         self.after_request_funcs = []
  27.     def add_prefix(self, prefix):
  28.         """add child path prefix
  29.         :param: prefix: node of url path
  30.         """
  31.         return Kotori(self.app, prefix, self)
  33.     @property
  34.     def path(self):
  35.         """get path of current node"""
  36.         if self._path is None:
  37.             x = [self.prefix]
  38.             k = self
  39.             while k.father is not None:
  40.                 x.append(k.father.prefix)
  41.                 k = k.father
  42.             self._path = ''.join(reversed(x))
  43.         return self._path
  45.     @setupmethod
  46.     def before_request(self, f):
  47.         """register a function for current node to run before each request"""
  48.         self.before_request_funcs.append(f)
  50.     @setupmethod
  51.     def after_request(self, f):
  52.         """register a function for current node to run after each request"""
  53.         self.after_request_funcs.append(f)
  55.     def _run_route_func(self, func_attribute_name):
  56.         funcs = []
  57.         current = self
  58.         while current.father is not None:
  59.             for f in getattr(current, func_attribute_name, []):
  60.                 funcs.append(f)
  61.             current = current.father
  62.         for f in reversed(funcs):
  63.             res = f()
  64.             if res is not None:
  65.                 return res
  67.     def _run_after_request(self):
  68.         return self._run_route_func('after_request_funcs')
  70.     def _run_before_request(self):
  71.         return self._run_route_func('before_request_funcs')
  73.     def route(self, rule, **options):
  74.         """a decorator that is used to register a view function for a given end-url-rule,
  75.         params are the same as Flask.route
  76.         """
  77.         rule = self.path + rule
  79.         def _(f):
  80.             @self.app.route(rule, **options)
  81.             @wraps(f)
  82.             def deco(*args, **kwargs):
  83.                 before_request_res = self._run_before_request()
  84.                 if before_request_res is not None:
  85.                     return before_request_res
  87.                 res = f(*args, **kwargs)
  89.                 after_request_res = self._run_after_request()
  90.                 if after_request_res is not None:
  91.                     return after_request_res
  92.                 return res
  93.             return deco
  94.         return _

使用方式如下：

  1. # coding: utf-8
  3. import json
  4. from flask import Flask
  5. from kotori import Kotori
  7. app = Flask(__name__)
  8. kotori_app = Kotori(app)
  9. manage_app = kotori_app.add_prefix('/manage')
  11. @app.before_request
  12. def before_request():
  13.     # load user
  16. service1 = manage_app.add_prefix('/service1')
  17. god_api = service1.add_prefix('/god/api')
  18. courses = god_api.add_prefix('/courses/<int:id>')
  22. @service1.before_request
  23. def service1_validator():
  24.     # do some validation
  25.     pass
  27. @courses.before_request
  28. def courses_validator():
  29.     # do some validation
  30.     pass
  33. @courses.route('/show.json')
  34. def service1_god_api_courses_all():
  35.     pass
  38. if __name__ == '__main__':
  39.     app.run(debug=True, port=9000)

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

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

言：
三尺书生，一介微命。

Blog

Thoughts

Project