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

2021-03-22 12:25:26 浏览数 (1)

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

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

前几天有人辗转找到公众号，留言询问之前一篇介绍 Flask-RESTPlus 文章的源代码（获得该文章请在公众号回复 swagger），Flask-RESTPlus 虽然看起来非常方便，但在实际编写代码时总有种和当前项目结构冲突的感觉，因此整理一篇改造某项目的总结，分享并探讨最佳实践。

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

蓝图与 API

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

代码语言：javascript复制
app = Flask(__name__)
api = Api(app)

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

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

代码语言：javascript复制
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 目录结构如下：

代码语言：javascript复制
.
├── __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 权限认证。在之前是这样处理的：

代码语言：javascript复制
@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 验证，我也将相关逻辑封装了继承的视图基类里，相关代码：

代码语言：javascript复制
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：

代码语言：javascript复制
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 也会在这派上用场：

代码语言：javascript复制
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

api flask json http 网站

0 人点赞