Flask 中的自定义身份验证和参数检查

Question

我目前正在重构我大约两年前编写的 Flask 应用程序，我怀疑我所做的一些事情不像使用该库那样优雅和干净。因此，我想就如何改善现状征求一些建议：

1. 该应用提供了多个 API 端点，每个端点都可以通过 /<category>/<endpoint> 形式的路由访问，其中 <category> 是 10 个不同类别之一。
2. 对于每个<category>，我创建了一个独特的Python模块<category>.py，将它放置在一个api/子目录和一个独特的flask.Blueprint（实际上是它的一个子类，见下文）并将其注册到主@ 987654327@.
3. 每个模块<category>.py 包含许多函数，它们充当各个类别的端点，并且每个都用路由装饰器进行装饰。

到目前为止一切顺利。

每个端点函数都接受许多参数，这些参数可以作为 GET 请求的参数传递，也可以作为 POST 请求的 JSON 有效负载中 parameter 字段的一部分传递。因此，在调用相应的端点函数之前，它会检查是否提供了正确数量的具有正确名称的参数。

此外，应用程序需要检查是否允许客户端调用某个函数。为此，读取由网络服务器设置的环境变量SSL_CLIENT_CERT（在我的例子中是通过 FCGI 的 lighttpd），并将其指纹与一些内部权限文件进行比较。

由于我不太清楚将上述逻辑放在哪里，我将flask.Blueprint 子类化并编写了我自己的（修改后的）route 装饰器（custom_route）。这个装饰器现在要么返回自定义错误响应（flask.Response 对象），要么返回自定义成功响应（从而使用从客户端传递的参数调用端点函数）。

所以模块category_xy.py 看起来像这样：

category_xy = CustomBlueprint('category_xy', __name__)    

@category_xy.custom_route('/category_xy/some_endpoint',
                          auth_required=True,
                          methods=['GET', 'POST'])
def some_endpoint(foo, bar):
    # do stuff 
    return '123'

CustomBlueprint 在单独的文件中定义为（部分伪代码）：

from flask import Blueprint, Response, g, request
from functools import wraps


class CustomBlueprint(Blueprint):
    def custom_route(self, rule, **options):
        def decorator(f):

            # don't pass the custom parameter 'auth_required' on to
            # self.route
            modified_options = dict(options)
            if 'auth_required' in modified_options:
                del modified_options['auth_required']

            @self.route(rule, **modified_options)
            @wraps(f)
            def wrapper():
                # do some authentication checks...
                if not authenticiated():
                    return Response(...)

                # check if correct paramters have been passed
                if not correct_paramters():
                    return Response(...)

                # extract parameter values from either GET or POST request
                arg_values = get_arg_values(request)

                # call the decorated function with these parameters
                result = f(*arg_values)

                return Response(result, ...)
            return wrapper
        return decorator

这行得通，但感觉一点也不干净，我认为应该有一种更好、更清洁的方法来做到这一点。将所有这些逻辑放在自定义装饰器中感觉非常错误。

有没有比 Flask 更有经验的人提供一些想法和/或最佳实践？

Answer 1

首先，您可以切换到Pluggable Views — 允许您将视图函数编写为类的 API：

from flask import Blueprint, Response, render_template
from flask.views import View

from models import db, CategoryXY # Let`s have here a database model

category_blueprint = Blueprint('category_xy', __name__)

class CategoryXYView(View):
    def get_list(self):
        items = db.session.query(CategoryXY).all()
        return render_template('category_xy_list.html', items=items)

    def get_one(self, item_id):
        item = db.session.query(CategoryXY).first()
        if item is None:
            return Response(status=404)
        return render_template('category_xy_edit.html', item=item)

    def dispatch_request(self, item_id):
        """Required method which handles incoming requests"""
        if item_id is None:
            return self.get_list()
        return self.get_one(item_id)

category_xy_view = CategoryXYView.as_view('category_xy')
category_blueprint.add_url_rule(
    'category_xy',
    view_func=category_xy_view,
    defaults={'item_id': None}
)
category_blueprint.add_url_rule(
    'category_xy/<item_id>',
    view_func=category_xy_view
)

当您拥有具有通用方法（数据库对象处理、验证、授权检查）的基视图类并且从它继承“类别”类时，它会变得非常有用：

class BaseView(View):
    model = None
    edit_template = None
    list_template = None

    def get_one(self, item_id):
        item = db.session.query(self.model).filter_by(id=item_id).first()
        if item is None:
            return Response(status=404)
        return render_template(self.edit_template, item=item)

    def get_list(self):
        items = db.session.query(self.model).all()
        return render_template(self.list_template, items=items)

    def dispatch_request(self, item_id):
        if item_id is None:
            return self.get_list()
        return self.get_one(item_id)

class CategoryXYView(BaseView):
    model = CategoryXY
    edit_template = 'category_xy_edit.html'
    list_template = 'category_xy_list.html'

category_xy_view = CategoryXYView.as_view('category_xy')
category_blueprint.add_url_rule(
    'category_xy',
    view_func=category_xy_view,
    defaults={'item_id': None}
)
category_blueprint.add_url_rule(
    'category_xy/<item_id>',
    view_func=category_xy_view
)

如果其中一个类别具有扩展功能，您只需覆盖其中一种常用方法：

class CategoryXXView(BaseView):
    model = CategoryXX
    edit_template = 'category_xx_edit.html'
    list_template = 'category_xx_list.html'

    def get_one(self, item_id):
        item = db.session.query(self.model).first()
        if item is None:
            return Response(status=404)
        xy_items = db.session.query(CategoryXY).all()
        return render_template(self.edit_template, item=item, xy_items=xy_items)

为了自动生成路由，您可以为BaseView 编写一个元类。每个新类的路由都会在类定义之后生成：

class ViewMeta(type):
    def __new__(mcs, name, bases, attributes):
        cls = type(name, bases, attributes)

        if cls.route is not None:
            view_function = cls.as_view(cls.route)
            category_blueprint.add_url_rule(
                cls.route,
                view_func=view_function,
                defaults={'item_id': None}
            )
            category_blueprint.add_url_rule(
                '{}/<item_id>'.format(cls.route),
                view_func=view_function
            )
        return cls

class BaseView(View):
    __metaclass__ = ViewMeta
    route = None
    model = None
    # Other common attributes

class CategoryXYView(BaseView):
    route = 'category_xy'
    model = CategoryXY