使用 Flask-RESTPlus 构建生产级应用

2018/05/26 16:52 pm posted in  Python 黑魔法

本文来自对某项目的实践总结,敏感信息已被隐藏或被 Resource 一词代替。

前几天有人辗转找到公众号,留言询问之前一篇介绍 Flask-RESTPlus 文章的源代码(Flask Api 文档管理与 Swagger 上手),Flask-RESTPlus 虽然看起来非常方便,但在实际编写代码时总有种和当前项目结构冲突的感觉,因此整理之前的一篇改造某项目的总结,分享并探讨最佳实践。

在生成 Swagger 文档上,Flask-RESTPlus 是比较常用的 flask 拓展,但引入该插件需要对项目结构些许调整,如果是从 0 到 1 的新项目,倒也无伤大雅,但是对于已经存在的旧项目,改造还是有一定的工作量的,本文通过总结具体的项目改造,对 Flask-RESTPlus 进一步的讲解,以此总结。

蓝图与 API

在大型 Flask 项目中,为了防止各个模块的依赖混乱,一般通过模块划分,并在 app 工厂方法中统一对各个模块的蓝图进行注册,Flask-RESTPlus 作为 flask 拓展可以通过与 flask app 绑定从而托管注册在 Flask-RESTPlus 的视图,比如官方文档的例子:

app = Flask(__name__)
api = Api(app)

但是这样会架空 flask 自带的蓝图,如果是新项目的话可以考虑使用 Flask-RESTPlus 的 Namespace 替代,但是如果是老项目迁移,成本还是蛮高的,因此可以将 蓝图与 Flask-RESTPlus Api 绑定,这样既保留了原有的模块划分,还可以利用 Namespace 进行更细致的逻辑划分。

比如对于当然项目来说,其中有多个 blueprint,来分割相对独立的模块,我们拿 Resource 模块举例,通过 flask 的蓝图对大模块进行划分之后,再通过 Namespace 对细节再次划分:

desc = """
resource type 1, type 2, type 3, type 4, type 5 api
"""

resource_blueprint = Blueprint("Resource", __name__)
api = Api(resource_blueprint, version='1.0', title='Resource info',
          description=desc)
api.add_namespace(resource_type_1_api)
api.add_namespace(resource_type_2_api)
api.add_namespace(resource_type_3_api)
api.add_namespace(resource_type_4_api)
api.add_namespace(resource_type_5_api)

Resource 支持五种不同的类型,虽然这几种类型的 api 同属在一个蓝图里,但是其本身相对独立,因此可以使用 Namespace 做更细致的区分,然后将这五个 namespace 注册到 api 里。因此 blueprints 目录结构如下:

.
├── __init__.py
├── action
│   ├── __init__.py
│   ├── apis.py
│   └── dto.py
├── health.py
├── json_schema.py
└── resource
    ├── __init__.py
    ├── type_1.py
    ├── type_2.py
    ├── type_3.py
    ├── type_4.py
    ├── type_5.py
    └── dto.py
    

参数检查与权限验证

解决了注册问题,还有部分公共设施需要修改,比如参数检查和 api 权限认证。在之前是这样处理的:

@resource_blueprint.route("/", methods=['POST'])
@internal_token_validator
@request_json_validator(SEND_TYPE_1_SCHEMA)
@tracing_span("post_type_1:type_1_api")
def op_type_1():
    pass

token 验证的逻辑写在 internal_token_validator 装饰器中,虽然 Flask-RESTPlus api 类支持注册装饰器,但是因为并不是所有的 api 都需要 token 认证,因此并不能直接注册在其中,但是有认证的 api 比例非常多,依然选择装饰器,那么装饰数量将要突破 6 个而且到处写一样的逻辑非常丑,因此我继承了 Flask-RESTPlus 视图类 Resource,并复写了 dispatch 函数,如果有方法需要 token 认证则动态将 internal_token_validator 装饰器放在 method_decorators 中,而后者会在 Flask-RESTPlus 处理视图方法时调用。

虽然 Flask-RESTPlus 提供了提供了参数验证的功能,但是对我们来讲并不够用(并不强大),而 DCS 中的参数验证一直使用的是 json-schema,在上面的例子中 request_json_validator 装饰器便是处理相关逻辑,该装饰器会将一个 json-schema 规则传入,然后在处理该 api 函数前将 request 中的 json body 验证,如果验证失败便会封装一个友好的 400 Response。

为了方便使用 json-schema 验证,我也将相关逻辑封装了继承的视图基类里,相关代码:

class BaseView(Resource):
    json_schemas = {}
    internal_token_required = ['get', 'post', 'delete', 'put', 'patch']

    def dispatch_request(self, *args, **kwargs):
        from message.common.errors import APIError
        try:
            method = request.method.lower()
            self.validate(self.json_schemas.get(method))
            if method in self.internal_token_required:
                self.method_decorators.append(internal_token_validator)
            return super(BaseView, self).dispatch_request(*args, **kwargs)
        except APIError as e:
            rsp = Response(
                response=json.dumps(e.render()), status=e.status_code,
                mimetype='application/json')
            return rsp

    @staticmethod
    def validate(schema):
        from message.common.errors import APIError
        if not schema:
            return
        data = request.json
        try:
            validate(data, schema)
        except ValidationError as e:
            raise APIError("ARGS_ERROR", e.message, 400)

DTO

最后谈一下导包的问题,在前一篇文章也提到 Flask-RESTPlus 容易产生相互引用, 而工程和 demo 不同,不能通过什么魔法技巧来避免这个问题 ,而应该通过更加细致的模块划分来避免,最后看到文章《How to structure a Flask-RESTPlus web service for production builds》(文后附链接)中介绍了 DTO 才让我找到了更 “结构化” 的解决办法。

DTO 即 data transfer object,这样设计的思路是和蓝图类似,传统 flask 应用中,在 app 工厂方法注册蓝图,而蓝图内的包相对独立,而 Flask-RESTPlus 引入了 namespace,按上文,我们把它作为蓝图更细以级的存在,因此,可以参考蓝图,将 namespace 的定义和依赖封装在一个类中,这样既避免了循环引用,还可以让整个项目的结构更清晰。

比如 Type1 DTO:

class Type1Dto:
    api = Namespace("type1", path="/type1", decorators=[internal_token_validator])
    action_model = api.model('ActionModel', action_desc)
    create_model = api.model("CreateType1Model", {
        "type1_title": fields.List(fields.String(description="Type1 title"), required=True),
        "type1_info": fields.Nested(api.model("Type1InfoModel", {
            "content": fields.String(description="Type1 content"),
        }), required=True),
    })
    template_model = api.model("TemplateType1Model", {
        "type1_title": fields.List(fields.String(description="Type1 title", required=True)),
        "content": fields.Nested(api.model("TemplateContent", {}), required=True),
    })
    model = api.model("Type1Model", {
        "total": fields.Integer(readOnly=True, description="action total"),
        "results": fields.List(fields.Nested(action_model)),
    })

其中包含了 namespace 的定义,request 的格式对象(Flask-RESTPlus 基于它生成 Request 文档),和 response 的返回对象(Flask-RESTPlus 基于它渲染 json 并生成 Response 文档)。

在使用时,将 dto 导入到视图层,而相关 model 也会在这派上用场:

from .dto import Type1Dto

api = Type1Dto.api


@api.route("/")
class Type1Api(Resource):
    json_schemas = {"post": SEND_TYPE_1_SCHEMA}

    @tracing_span("post_type_1:type_1_api")
    @api.expect(Type1Dto.create_model)
    @api.param("Token", description="internal token.", _in="header", required=True)
    def post(self):
        actions = deal_with_type_1(api.payload['type_1_title'], api.payload['content'])
        result = {
            "total": len(actions),
            "results": [a.render() for a in actions]
        }
        return result

最后将视图层的 api 导入到蓝图定义的地方完成注册,这样整个项目既做到了合理的结构分类,也完成和解决了导包问题。


参考资料:《How to structure a Flask-RESTPlus web service for production builds》: https://medium.freecodecamp.org/structuring-a-flask-restplus-web-service-for-production-builds-c2ec676de563