写一个 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 真好使。