Flask-RestX 概述

Flask-RESTX 是针对 Flask 的扩展，它增加了对快速构建 REST API 的支持。
Flask-RESTX 希望能通过尽可能少的代码来完成相关的各种配置。如果您熟悉 Flask，Flask-RESTX 应该很容易上手。
具体来说，它提供了一系列装饰器和工具来描述您的 API 并能自动生成 Swagger 并对外提供。

安装

Flask-RestX 的安装非常简单，可以直接使用 pip 包管理工具安装即可：

  1. pip install flask-restx

快速上手

首先，我们来看一个最简的 Flask-RestX 的应用是什么样的！

  1. from flask import Flask
  2. from flask_restx import Resource, Api
  4. app = Flask(__name__)
  5. api = Api(app)
  7. @api.route('/hello')
  8. class HelloWorld(Resource):
  9.     def get(self):
  10.         return {'hello': 'world'}
  12. if __name__ == '__main__':
  13.     app.run(debug=True)

首先，可以看到相比原始的 Flask 程序而言，我们修改了哪些内容：

  • 将 Flask app 用 Api 来进行了装饰。
  • url 的注释从 app 修改为了 Api 装饰后的 api 对象。
  • View 中仅支持了 Class，且该 Class 需要继承自 Resource 类。
  • View 类中的方法名称对应于 HTTP 协议中的 method。

此时，再次启动服务后，你不但能够访问 /hello API，同时还可以在/根目录看到对应的 Swagger 的接口文档页面。
image.png

资源路由

Flask-RESTX 提供的主要构建基础是 Resource，Flask-RestX 支持多种不同资源请求方式的方式也很简单，就是在 Class 中定义多个不同的请求方法名称即可，示例如下：

  1. from flask import Flask, request
  2. from flask_restx import Resource, Api
  4. app = Flask(__name__)
  5. api = Api(app)
  7. todos = {}
  9. @api.route('/<string:todo_id>')
  10. class TodoSimple(Resource):
  11.     def get(self, todo_id):
  12.         return {todo_id: todos[todo_id]}
  14.     def put(self, todo_id):
  15.         todos[todo_id] = request.form['data']
  16.         return {todo_id: todos[todo_id]}
  18. if __name__ == '__main__':
  19.     app.run(debug=True)

此外，在对应的 ViewClass 中方法中，Flask-RESTX 支持根据返回值的个数自动判断返回响应的内容：

  1. class Todo1(Resource):
  2.     def get(self):
  3.         # Default to 200 OK
  4.         return {'task': 'Hello world'}
  6. class Todo2(Resource):
  7.     def get(self):
  8.         # Set the response code to 201
  9.         return {'task': 'Hello world'}, 201
  11. class Todo3(Resource):
  12.     def get(self):
  13.         # Set the response code to 201 and return custom headers
  14.         return {'task': 'Hello world'}, 201, {'Etag': 'some-opaque-string'}

其中，可以返回响应体、响应码以及响应Headers。

入口URL

很多时候，我们希望能有多个不同的 URL 访问到同一个资源上。这时，我们可以将多个 URL 传递给 Api 对象上的 add_resource() 方法或 route() 装饰器。这样，每个url 都将路由到我们的资源上：

  1. api.add_resource(HelloWorld, '/hello', '/world')
  3. # or
  5. @api.route('/hello', '/world')
  6. class HelloWorld(Resource):
  7.     pass

此外，我们还可以将路径的一部分作为变量匹配到您的资源方法：

  1. api.add_resource(Todo, '/todo/<int:todo_id>', endpoint='todo_ep')
  3. # or
  5. @api.route('/todo/<int:todo_id>', endpoint='todo_ep')
  6. class HelloWorld(Resource):
  7.     pass

参数解析

虽然 Flask 本文已经提供了对请求数据（即查询字符串或 POST 表单编码数据）的轻松访问的机制，但验证表单数据是否合法其实还是很复杂的。而在 Flask-RESTX 内置了对使用类似于 argparse 的库的请求数据验证的支持。
一个简单的示例如下：

  1. from flask_restx import reqparse
  3. parser = reqparse.RequestParser()
  4. parser.add_argument('rate', type=int, help='Rate to charge for this resource')
  5. args = parser.parse_args()

其中，parse_args() 返回的数据类型是 dict。
除了校验之外，使用 RequestParser 类还可以自动为您提供相同格式的错误消息。如果参数未能通过验证，Flask-RESTX 将响应 400 Bad Request 和显示错误的原因提示信息。

  1. curl -d 'rate=foo' http://127.0.0.1:5000/todo/123
  2. # {"errors": {"rate": "Rate to charge for this resource invalid literal for int() with base 10: 'foo'"}, "message": "Input payload validation failed"}

此外，输入模块还提供了许多包含的常用转换函数，例如 date() 和 url()。
如果在调用 parse_args() 方法时，传入 strict=True 的参数时，此时如果请求包含您的解析器未定义的参数，则会引发错误。

数据格式化

默认在 Flask 中，返回对象中的所有字段都将原样作为HTTP响应信息返回。这样对于处理 Python 的数据结构而言，本身是非常方便的，但是在返回的是 Python 对象时，则处理起来会非常的复杂。
为了解决这个问题，Flask-RESTX 提供了 fields 模块和 marshal_with() 装饰器，它们的功能与 Django ORM 和 WTForm 的功能类似，可以使用 fields 模块来描述我们的响应结构。

一个简单的示例代码如下所示：

  1. from flask import Flask
  2. from flask_restx import Resource, Api, fields
  4. app = Flask(__name__)
  5. api = Api(app)
  8. model = api.model('Model', {
  9.     'task': fields.String,
  10.     'uri': fields.Url('todo')
  11. })
  14. class TodoDao(object):
  15.     def __init__(self, todo_id, task):
  16.         self.todo_id = todo_id
  17.         self.task = task
  19.         # This field will not be sent in the response
  20.         self.status = 'active'
  23. @api.route('/todo')
  24. class Todo(Resource):
  25.     @api.marshal_with(model)
  26.     def get(self, **kwargs):
  27.         return TodoDao(todo_id='my_todo', task='Remember the milk')
  30. if __name__ == '__main__':
  31.     app.run(debug=True)

上面的示例采用一个 python 对象作为返回值进行返回，并在返回过程中将其序列化。
marshal_with() 装饰器将应用模型描述的转换。从对象中提取的唯一字段是 task。 fields.Url 字段是一个特殊字段，它采用url名称并在响应中为该生成对应的 URL。使用 marshal_with() 装饰器还可以在 swagger 规范中记录对应的输出格式。

最后，我们来看一个完整的示例项目代码吧：

  1. from flask import Flask
  2. from flask_restx import Api, Resource, fields
  3. from werkzeug.middleware.proxy_fix import ProxyFix
  5. app = Flask(__name__)
  6. app.wsgi_app = ProxyFix(app.wsgi_app)
  7. api = Api(app, version='1.0', title='TodoMVC API',
  8.     description='A simple TodoMVC API',
  9. )
  11. ns = api.namespace('todos', description='TODO operations')
  13. todo = api.model('Todo', {
  14.     'id': fields.Integer(readonly=True, description='The task unique identifier'),
  15.     'task': fields.String(required=True, description='The task details')
  16. })
  19. class TodoDAO(object):
  20.     def __init__(self):
  21.         self.counter = 0
  22.         self.todos = []
  24.     def get(self, id):
  25.         for todo in self.todos:
  26.             if todo['id'] == id:
  27.                 return todo
  28.         api.abort(404, "Todo {} doesn't exist".format(id))
  30.     def create(self, data):
  31.         todo = data
  32.         todo['id'] = self.counter = self.counter + 1
  33.         self.todos.append(todo)
  34.         return todo
  36.     def update(self, id, data):
  37.         todo = self.get(id)
  38.         todo.update(data)
  39.         return todo
  41.     def delete(self, id):
  42.         todo = self.get(id)
  43.         self.todos.remove(todo)
  46. DAO = TodoDAO()
  47. DAO.create({'task': 'Build an API'})
  48. DAO.create({'task': '?????'})
  49. DAO.create({'task': 'profit!'})
  52. @ns.route('/')
  53. class TodoList(Resource):
  54.     '''Shows a list of all todos, and lets you POST to add new tasks'''
  55.     @ns.doc('list_todos')
  56.     @ns.marshal_list_with(todo)
  57.     def get(self):
  58.         '''List all tasks'''
  59.         return DAO.todos
  61.     @ns.doc('create_todo')
  62.     @ns.expect(todo)
  63.     @ns.marshal_with(todo, code=201)
  64.     def post(self):
  65.         '''Create a new task'''
  66.         return DAO.create(api.payload), 201
  69. @ns.route('/<int:id>')
  70. @ns.response(404, 'Todo not found')
  71. @ns.param('id', 'The task identifier')
  72. class Todo(Resource):
  73.     '''Show a single todo item and lets you delete them'''
  74.     @ns.doc('get_todo')
  75.     @ns.marshal_with(todo)
  76.     def get(self, id):
  77.         '''Fetch a given resource'''
  78.         return DAO.get(id)
  80.     @ns.doc('delete_todo')
  81.     @ns.response(204, 'Todo deleted')
  82.     def delete(self, id):
  83.         '''Delete a task given its identifier'''
  84.         DAO.delete(id)
  85.         return '', 204
  87.     @ns.expect(todo)
  88.     @ns.marshal_with(todo)
  89.     def put(self, id):
  90.         '''Update a task given its identifier'''
  91.         return DAO.update(id, api.payload)
  94. if __name__ == '__main__':
  95.     app.run(debug=True)

响应格式化

在快速上手中，我们已经了解到了 Flask-RESTX 提供了一种机制，可以将你的响应结构按照预期的格式进行格式化，其中主要借助的就是 fields 模块，通过该模块，我们可以定义我们响应的数据格式，同时不会暴露出内部的数据结构。

基础用法

你可以定义一个字典（dict）或有序字典（OrderedDict）来存放其键的属性名称或要渲染的对象上的键的字段，并且其值是将格式化并返回该字段的值的类。
下面的例子有三个字段：2个是 String ，1个是 DateTime ， DateTime 将被格式化为ISO 8601日期时间字符串：

  1. from flask_restx import Resource, fields
  3. model = api.model('Model', {
  4.     'name': fields.String,
  5.     'address': fields.String,
  6.     'date_updated': fields.DateTime(dt_format='rfc822'),
  7. })
  9. @api.route('/todo')
  10. class Todo(Resource):
  11.     @api.marshal_with(model, envelope='resource')
  12.     def get(self, **kwargs):
  13.         return db_get_todo()

在这个例子里面你有一个自定义的数据库对象（todo），它拥有 name 、 address 和 date_updated 属性。整个对象的其他附加属性都将是私有的并且不会在输出中渲染。
此外，我们指定了一个 envelope 关键字参数来包装结果输出。

字段重命名

很多时候内部字段名和外部的字段名是不一样的。你可以使用 attribute 关键字来配置这种映射关系：

  1. model = {
  2.     'name': fields.String(attribute='private_name'),
  3.     'address': fields.String,
  4. }

嵌套属性（Nested properties）也可以通过 attribute 进行访问:

  1. model = {
  2.     'name': fields.String(attribute='people_list.0.person_dictionary.name'),
  3.     'address': fields.String,
  4. }

默认值

如果你的数据对象里没有字段列表中对应的属性，你可以指定一个默认值而不是返回一个 None：

  1. model = {
  2.     'name': fields.String(default='Anonymous User'),
  3.     'address': fields.String,
  4. }

复杂结构

你可以通过 marshal() 函数将平面的数据结构转化成嵌套的数据结构：

  1. >>> from flask_restx import fields, marshal
  2. >>> import json
  3. >>>
  4. >>> resource_fields = {'name': fields.String}
  5. >>> resource_fields['address'] = {}
  6. >>> resource_fields['address']['line 1'] = fields.String(attribute='addr1')
  7. >>> resource_fields['address']['line 2'] = fields.String(attribute='addr2')
  8. >>> resource_fields['address']['city'] = fields.String
  9. >>> resource_fields['address']['state'] = fields.String
  10. >>> resource_fields['address']['zip'] = fields.String
  11. >>> data = {'name': 'bob', 'addr1': '123 fake street', 'addr2': '', 'city': 'New York', 'state': 'NY', 'zip': '10468'}
  12. >>> json.dumps(marshal(data, resource_fields))
  13. '{"name": "bob", "address": {"line 1": "123 fake street", "line 2": "", "state": "NY", "zip": "10468", "city": "New York"}}'

字段列表

你也可以将字段解组成列表：

  1. >>> from flask_restx import fields, marshal
  2. >>> import json
  3. >>>
  4. >>> resource_fields = {'name': fields.String, 'first_names': fields.List(fields.String)}
  5. >>> data = {'name': 'Bougnazal', 'first_names' : ['Emile', 'Raoul']}
  6. >>> json.dumps(marshal(data, resource_fields))
  7. >>> '{"first_names": ["Emile", "Raoul"], "name": "Bougnazal"}'

通配符字段

如果你不知道你要解组的字段名称，你可以使用 通配符（Wildcard） ：

  1. >>> from flask_restx import fields, marshal
  2. >>> import json
  3. >>>
  4. >>> wild = fields.Wildcard(fields.String)
  5. >>> wildcard_fields = {'*': wild}
  6. >>> data = {'John': 12, 'bob': 42, 'Jane': '68'}
  7. >>> json.dumps(marshal(data, wildcard_fields))
  8. >>> '{"Jane": "68", "bob": "42", "John": "12"}'

通配符 的名称将作为匹配的依据，如下所示:

  1. >>> from flask_restx import fields, marshal
  2. >>> import json
  3. >>>
  4. >>> wild = fields.Wildcard(fields.String)
  5. >>> wildcard_fields = {'j*': wild}
  6. >>> data = {'John': 12, 'bob': 42, 'Jane': '68'}
  7. >>> json.dumps(marshal(data, wildcard_fields))
  8. >>> '{"Jane": "68", "John": "12"}'

为了避免意料之外的情况，在混合 通配符 字段和其他字段使用的时候，请使用 有序字典(OrderedDict) 并且将 通配符 字段放在最后面:

  1. >>> from flask_restx import fields, marshal
  2. >>> from collections import OrderedDict
  3. >>> import json
  4. >>>
  5. >>> wild = fields.Wildcard(fields.Integer)
  6. >>> mod = OrderedDict()
  7. >>> mod['zoro'] = fields.String
  8. >>> mod['*'] = wild
  9. >>> # you can use it in api.model like this:
  10. >>> # some_fields = api.model('MyModel', mod)
  11. >>>
  12. >>> data = {'John': 12, 'bob': 42, 'Jane': '68', 'zoro': 72}
  13. >>> json.dumps(marshal(data, mod))
  14. >>> '{"zoro": "72", "Jane": 68, "bob": 42, "John": 12}'

嵌套字段

虽然你可以通过嵌套字段将平面数据转换成多层结构的响应，但是你也可以通过 Nested 将多层结构的数据转换成适当的形式：

  1. >>> from flask_restx import fields, marshal
  2. >>> import json
  3. >>>
  4. >>> address_fields = {}
  5. >>> address_fields['line 1'] = fields.String(attribute='addr1')
  6. >>> address_fields['line 2'] = fields.String(attribute='addr2')
  7. >>> address_fields['city'] = fields.String(attribute='city')
  8. >>> address_fields['state'] = fields.String(attribute='state')
  9. >>> address_fields['zip'] = fields.String(attribute='zip')
  10. >>>
  11. >>> resource_fields = {}
  12. >>> resource_fields['name'] = fields.String
  13. >>> resource_fields['billing_address'] = fields.Nested(address_fields)
  14. >>> resource_fields['shipping_address'] = fields.Nested(address_fields)
  15. >>> address1 = {'addr1': '123 fake street', 'city': 'New York', 'state': 'NY', 'zip': '10468'}
  16. >>> address2 = {'addr1': '555 nowhere', 'city': 'New York', 'state': 'NY', 'zip': '10468'}
  17. >>> data = {'name': 'bob', 'billing_address': address1, 'shipping_address': address2}
  18. >>>
  19. >>> json.dumps(marshal(data, resource_fields))
  20. '{"billing_address": {"line 1": "123 fake street", "line 2": null, "state": "NY", "zip": "10468", "city": "New York"}, "name": "bob", "shipping_address": {"line 1": "555 nowhere", "line 2": null, "state": "NY", "zip": "10468", "city": "New York"}}'

这个例子使用了两个 嵌套字段（Nested fields） 。嵌套字段 的构造函数需要一个字段字典作为子字段的输入。
一个 嵌套字段（Nested） 构造器和之前嵌套字典（之前的例子）的区别是：属性的上下文。在这个例子中，billing_address 是一个拥有子字段的复杂对象并且传递给嵌套字段的上下文是子对象，而不是原始 data 对象。
换句话说： data.billing_address.addr1 作用域在这，而之前例子中 data.addr1 是本地（localtion）属性。请记住：嵌套字段（Nested） 和 列表字段（List） 会为属性创建新的作用域。
使用 嵌套字段 和 列表字段 来编组拥有复杂结构的列表对象：

  1. user_fields = api.model('User', {
  2.     'id': fields.Integer,
  3.     'name': fields.String,
  4. })
  6. user_list_fields = api.model('UserList', {
  7.     'users': fields.List(fields.Nested(user_fields)),
  8. })

使用JSON格式定义模型

在 Flask-RESTX 中，支持使用 JSON Schema 来定义数据库模型，示例如下：

  1. address = api.schema_model('Address', {
  2.     'properties': {
  3.         'road': {
  4.             'type': 'string'
  5.         },
  6.     },
  7.     'type': 'object'
  8. })
  10. person = address = api.schema_model('Person', {
  11.     'required': ['address'],
  12.     'properties': {
  13.         'name': {
  14.             'type': 'string'
  15.         },
  16.         'age': {
  17.             'type': 'integer'
  18.         },
  19.         'birthdate': {
  20.             'type': 'string',
  21.             'format': 'date-time'
  22.         },
  23.         'address': {
  24.             '$ref': '#/definitions/Address',
  25.         }
  26.     },
  27.     'type': 'object'
  28. })

请求参数解析

基于Flask-RESTful的请求解析接口 reqparse 是在argparse接口之上扩展的。
它的设计提供了简单和统一的访问 flask.request 对象上的任何变量。

基础参数

下面是请求解析器的一个简单示例。它在flask.Request.values字典中查找两个参数:一个整数和一个字符串：

  1. from flask_restx import reqparse
  3. parser = reqparse.RequestParser()
  4. parser.add_argument('rate', type=int, help='Rate cannot be converted')
  5. parser.add_argument('name')
  6. args = parser.parse_args()

必填参数

要为参数设置为必填，只需将required=True参数添加到add_argument()调用即可：

  1. parser.add_argument('name', required=True, help="Name cannot be blank!")

列表参数

如果您希望一个键能以列表的形式接收多个值，可以像下面这样传递参数action=’append’：

  1. parser.add_argument('name', action='append')

如果您希望用逗号分隔的字符串能被分割成列表，并作为一个键的值，可以像下面这样传递参数action=’split’:

  1. parser.add_argument('fruits', action='split')

重命名

如果出于某种原因，希望在解析后将参数存储在不同的名称下，那么可以使用dest关键字参数：

  1. parser.add_argument('name', dest='public_name')
  3. args = parser.parse_args()
  4. args['public_name']

参数传入位置

默认情况下，RequestParser尝试解析来自flask.Request.values和flask.Request.json的值。
使用add_argument()的location参数来指定从哪些位置获取值。flask.Request上的任何变量都可以使用。例如：

  1. # Look only in the POST body
  2. parser.add_argument('name', type=int, location='form')
  3. # Look JSON body
  4. parser.add_argument('name', type=int, location='json')
  5. # Look only in the querystring
  6. parser.add_argument('PageSize', type=int, location='args')
  7. # From the request headers
  8. parser.add_argument('User-Agent', location='headers')
  9. # From http cookies
  10. parser.add_argument('session_id', location='cookies')
  11. # From file uploads
  12. parser.add_argument('picture', type=werkzeug.datastructures.FileStorage, location='files')

多位置参数

可以通过将列表传递给location来指定多个参数位置：

  1. parser.add_argument('text', location=['headers', 'values'])

当指定多个位置时，来自指定的所有位置的参数将合并为单个MultiDict。最后列出的位置优先于结果集。
如果参数位置列表包含headers，则参数名将区分大小写，必须与标题大小写名称完全匹配。

类型校验

有时，您需要比基本数据类型更多的类型来处理输入验证。input模块提供一些常见的类型处理：

  • 用于更加广泛布尔处理的 boolean()
  • 用于IP地址的 ipv4() 和 ipv6()
  • 用于ISO8601日期和数据处理的date_from_iso8601() 和 datetime_from_iso8601()

你只需要把它们用在 type 参数上：

  1. parser.add_argument('flag', type=inputs.boolean)

关于 input 的支持的完整列表，可以参考相关文档。
此外，也可以自定编写自己的数据类型：

  1. def my_type(value):
  2.     '''Parse my type'''
  3.     if not condition:
  4.         raise ValueError('This is not my type')
  5.     return parse(value)
  7. # Swagger documntation
  8. my_type.__schema__ = {'type': 'string', 'format': 'my-custom-format'}

解析器继承

通常地，您会为您写的每一份资源配置不同的解析器。这样做的问题会造成大量的重复。
我们可以看一下解析器是否具有公共的参数，不同于重新写参数，您可以编写一个包含所有公共的参数的父解析器，然后用 copy()函数继承这个解析器。您也可以用 replace_argument()来重写父解析器里的任何参数，或者干脆用 remove_argument() 完全移除它。

  1. from flask_restx import reqparse
  3. parser = reqparse.RequestParser()
  4. parser.add_argument('foo', type=int)
  6. parser_copy = parser.copy()
  7. parser_copy.add_argument('bar', type=int)
  9. # parser_copy has both 'foo' and 'bar'
  11. parser_copy.replace_argument('foo', required=True, location='json')
  12. # 'foo' is now a required str located in json, not an int as defined
  13. #  by original parser
  15. parser_copy.remove_argument('foo')
  16. # parser_copy no longer has 'foo' argument

文件上传

要使用 RequestParser 处理文件上传，您需要使用 files 位置并将 type设置为FileStorage。

  1. from werkzeug.datastructures import FileStorage
  3. upload_parser = api.parser()
  4. upload_parser.add_argument('file', location='files',
  5.                            type=FileStorage, required=True)
  8. @api.route('/upload/')
  9. @api.expect(upload_parser)
  10. class Upload(Resource):
  11.     def post(self):
  12.         uploaded_file = args['file']  # This is FileStorage instance
  13.         url = do_something_with_file(uploaded_file)
  14.         return {'url': url}, 201

错误处理

RequestParser 处理错误的默认方式是在第一个错误出现的时候终止，然而，将错误捆绑一起并且一次性发送回客户端通常是较好的处理。
这种方式可以在Flask应用程序级别（Flask application level）或在特定的 RequestParser实例上被指定。为了在调用 RequestParser的时候使用捆绑错误的选项，需要传递 bundle_errors 参数。下面是一个示例：

  1. from flask_restx import reqparse
  3. parser = reqparse.RequestParser(bundle_errors=True)
  4. parser.add_argument('foo', type=int, required=True)
  5. parser.add_argument('bar', type=int, required=True)
  7. # If a request comes in not containing both 'foo' and 'bar', the error that
  8. # will come back will look something like this.
  10. {
  11.     "message":  {
  12.         "foo": "foo error message",
  13.         "bar": "bar error message"
  14.     }
  15. }
  17. # The default behavior would only return the first error
  19. parser = RequestParser()
  20. parser.add_argument('foo', type=int, required=True)
  21. parser.add_argument('bar', type=int, required=True)
  23. {
  24.     "message":  {
  25.         "foo": "foo error message"
  26.     }
  27. }

程序级别的配置是 “BUNDLE_ERRORS”，示例如下：

  1. from flask import Flask
  3. app = Flask(__name__)
  4. app.config['BUNDLE_ERRORS'] = True

错误提示

每个域的错误消息可以通过 help 参数来进行定制（也是在 RequestParser.add_argument当中）。
如果不提供 help 参数，那么这个域的错误消息会是错误类型本身的字符串表示。否则，错误消息就是 help 参数的值。
help 参数可能包含一个插值标记（ interpolation token），就像 {error_msg} 这样，这个标记将会被错误类型的字符串表示替换。这允许您在保留原本的错误消息的同时定制消息，就像下面的例子这样：

  1. from flask_restx import reqparse
  4. parser = reqparse.RequestParser()
  5. parser.add_argument(
  6.     'foo',
  7.     choices=('one', 'two'),
  8.     help='Bad choice: {error_msg}'
  9. )
  11. # If a request comes in with a value of "three" for `foo`:
  13. {
  14.     "message":  {
  15.         "foo": "Bad choice: three is not a valid choice",
  16.     }
  17. }

字段掩码

Flask-RESTPlus通过一个自定义请求头支持部分对象获取（partial object fetching）也就是字段掩码（fields mask）。默认情况下头名为 X-Fields ，但是它可以被 RESTPLUS_MASK_HEADER 参数修改。

语法

语法非常简单。你只要提供一个包含字段名并用逗号分隔的列表，可以选择性的用括号包裹：

  1. # These two mask are equivalents
  2. mask = '{name,age}'
  3. # or
  4. mask = 'name,age'
  5. data = requests.get('/some/url/', headers={'X-Fields': mask})
  6. assert len(data) == 2
  7. assert 'name' in data
  8. assert 'age' in data

实现嵌套字段只需要将内容用大括号包裹即可:

  1. mask = '{name, age, pet{name}}'

PS：嵌套规范适用于嵌套对象或对象列表。
特殊字符米字星（*）代表“所有剩余字段（all remaining fields）”。它仅允许指定嵌套过滤：

  1. # Will apply the mask {name} to each pet
  2. # in the pets list and take all other root fields
  3. # without filtering.
  4. mask = '{pets{name},*}'
  6. # Will not filter anything
  7. mask = '*'

默认情况下，每次使用 api.marshal 或 @api.marshal_with 时，如果存在标头，掩码将自动应用。
当你每次使用 @api.marshal_with 修饰器时，标头将作为Swagger参数展示。
你可以修改 RESTX_MASK_SWAGGER 为 False 来禁用这个功能。

Swagger 文档生成

Swagger API文档是自动生成的，可从您API的根目录访问，可以通过 doc=”” 的参数来修改对应的访问路径，示例如下：

  1. api = Api(app, version='1.0', title='TodoMVC API',
  2.     description='A simple TodoMVC API', doc='/doc/'
  3. )

使用@api.doc()装饰器进行文档编辑

@api.doc() 修饰器使你可以为文档添加额外信息。
你可以为一个类或者函数添加文档：

  1. @api.route('/my-resource/<id>', endpoint='my-resource')
  2. @api.doc(params={'id': 'An ID'})
  3. class MyResource(Resource):
  4.     def get(self, id):
  5.         return {}
  7.     @api.doc(responses={403: 'Not Authorized'})
  8.     def post(self, id):
  9.         api.abort(403)

模型自动记录

所有由 model() 、 clone() 和 inherit() 实例化的模型（model）都会被自动记录到Swagger文档中。
inherit() 函数会将父类和子类都注册到Swagger的模型（model）定义中：

  1. parent = api.model('Parent', {
  2.     'name': fields.String,
  3.     'class': fields.String(discriminator=True)
  4. })
  6. child = api.inherit('Child', parent, {
  7.     'extra': fields.String
  8. })

上面的代码会生成下面的Swagger定义：

  1. {
  2.     "Parent": {
  3.         "properties": {
  4.             "name": {"type": "string"},
  5.             "class": {"type": "string"}
  6.         },
  7.         "discriminator": "class",
  8.         "required": ["class"]
  9.     },
  10.     "Child": {
  11.         "allOf": [
  12.             {
  13.                 "$ref": "#/definitions/Parent"
  14.             }, {
  15.                 "properties": {
  16.                     "extra": {"type": "string"}
  17.                 }
  18.             }
  19.         ]
  20.     }
  21. }

@api.marshal_with()装饰器

这个装饰器和原生的 marshal_with() 装饰器几乎完全一致，除了这个装饰器会记录函数文档。可选参数代码允许您指定期望的HTTP状态码（默认为200）。可选参数 as_list 允许您指定是否将对象作为列表返回。

  1. resource_fields = api.model('Resource', {
  2.     'name': fields.String,
  3. })
  5. @api.route('/my-resource/<id>', endpoint='my-resource')
  6. class MyResource(Resource):
  7.     @api.marshal_with(resource_fields, as_list=True)
  8.     def get(self):
  9.         return get_objects()
  11.     @api.marshal_with(resource_fields, code=201)
  12.     def post(self):
  13.         return create_object(), 201

@api.expect()装饰器

@api.expect() 装饰器允许你指定所需的输入字段。它接受一个可选的布尔参数 validate ，指示负载（payload）是否需要被验证。
可以通过将 RESTPLUS_VALIDATE 配置设置为 True 或将 validate = True 传递给API构造函数来全局设置验证功能。
示例如下：

  1. resource_fields = api.model('Resource', {
  2.     'name': fields.String,
  3. })
  5. @api.route('/my-resource/<id>')
  6. class MyResource(Resource):
  7.     @api.expect(resource_fields)
  8.     def get(self):
  9.         pass

我们也可以将列表指定为预期输入：

  1. resource_fields = api.model('Resource', {
  2.     'name': fields.String,
  3. })
  5. @api.route('/my-resource/<id>')
  6. class MyResource(Resource):
  7.     @api.expect([resource_fields])
  8.     def get(self):
  9.         pass

此外，还可以使用 RequestParser 定义期望的输入：

  1. parser = api.parser()
  2. parser.add_argument('param', type=int, help='Some param', location='form')
  3. parser.add_argument('in_files', type=FileStorage, location='files')
  6. @api.route('/with-parser/', endpoint='with-parser')
  7. class WithParserResource(restplus.Resource):
  8.     @api.expect(parser)
  9.     def get(self):
  10.         return {}

一个完整示例如下：

  1. api = Api(app, validate=True)
  3. resource_fields = api.model('Resource', {
  4.     'name': fields.String,
  5. })
  7. @api.route('/my-resource/<id>')
  8. class MyResource(Resource):
  9.     # Payload validation enabled
  10.     @api.expect(resource_fields)
  11.     def post(self):
  12.         pass
  14.     # Payload validation disabled
  15.     @api.expect(resource_fields, validate=False)
  16.     def post(self):
  17.         pass

使用@api.response装饰器进行文档编辑

@api.response 装饰器允许您记录已知的响应，并且他是 @api.doc(responses=’…’) 的简化写法。
例如：

  1. @api.route('/my-resource/')
  2. class MyResource(Resource):
  3.     @api.doc(responses={
  4.         200: 'Success',
  5.         400: 'Validation Error'
  6.     })
  7.     def get(self):
  8.         pass

您可以选择将响应模型指定为第三个输入参数：

  1. model = api.model('Model', {
  2.     'name': fields.String,
  3. })
  5. @api.route('/my-resource/')
  6. class MyResource(Resource):
  7.     @api.response(200, 'Success', model)
  8.     def get(self):
  9.         pass

使用 @api.marshal_with() 修饰器会自动在 Swagger 记录响应文档：

  1. model = api.model('Model', {
  2.     'name': fields.String,
  3. })
  5. @api.route('/my-resource/')
  6. class MyResource(Resource):
  7.     @api.response(400, 'Validation error')
  8.     @api.marshal_with(model, code=201, description='Object created')
  9.     def post(self):
  10.         pass

如果你不知道状态码的时候，你可以指定一个默认响应：

  1. @api.route('/my-resource/')
  2. class MyResource(Resource):
  3.     @api.response('default', 'Error')
  4.     def get(self):
  5.         pass

@api.route()装饰器

你可以指定一个类范围的文档通过 Api.route() 的 doc 参数。这个参数可以接受和 @api.doc() 装饰器相同的参数。
例如：

  1. @api.route('/my-resource/<id>', endpoint='my-resource', doc={'params':{'id': 'An ID'}})
  2. class MyResource(Resource):
  3.     def get(self, id):
  4.         return {}

字段格式约束设置

每个Flask-RESTX字段都可以可选的接受以下几个用于文档的参数：

  • required ：一个Bool值标识这个字段是否是必要的（默认：False）
  • description：一些字段的详细描述（默认：None ）
  • example ：展示时显示的示例（默认： None ）

这里还有一些特定字段的属性：

  • String 字段可以接受以下参数：
    • enum ：一个限制授权值的数组。
    • min_length ：字符串的最小长度。
    • max_length ：字符串的最大长度。
    • pattern ：一个正则表达式用于验证。
  • Integer 、 Float 和 Arbitrary 字段可以接受以下参数：
    • min ：可以接受的最小值。
    • max ：可以接受的最大值。
    • ExclusiveMin ：如果为 True ，则区间不包含最小值。
    • exclusiveMax ：如果为 True ，则区间不包含最大值。
    • multiple ：限制输入的数字是这个数的倍数。
  • DateTime 字段可以接受 min 、 max 、 exclusiveMin 和 exclusiveMax 参数。这些参数必须是 dates 或 datetimes。

示例如下：

  1. my_fields = api.model('MyModel', {
  2.     'name': fields.String(description='The name', required=True),
  3.     'type': fields.String(description='The object type', enum=['A', 'B']),
  4.     'age': fields.Integer(min=0),
  5. })

函数文档编写

每一个资源类都会被记录为一个Swagger路径。
每一个资源类的函数（get , post , put , delete , path , options , head）都会被记录为一个Swagger操作。
示例如下：

  1. @api.route('/my-resource/')
  2. class MyResource(Resource):
  3.     @api.doc('get_something')
  4.     def get(self):
  5.         return {}

函数参数

来自url地址的参数会自动被记录。你可以通过 api.doc() 修饰器的 params 参数添加额外的内容：

  1. @api.route('/my-resource/<id>', endpoint='my-resource')
  2. @api.doc(params={'id': 'An ID'})
  3. class MyResource(Resource):
  4.     pass

可以通过 api.doc() 修饰器的 model 关键字指定序列化输出模型。
对于 PUT 和 POST 方法，使用 body 关键字指定输入模型。

  1. fields = api.model('MyModel', {
  2.     'name': fields.String(description='The name', required=True),
  3.     'type': fields.String(description='The object type', enum=['A', 'B']),
  4.     'age': fields.Integer(min=0),
  5. })
  8. @api.model(fields={'name': fields.String, 'age': fields.Integer})
  9. class Person(fields.Raw):
  10.     def format(self, value):
  11.         return {'name': value.name, 'age': value.age}
  14. @api.route('/my-resource/<id>', endpoint='my-resource')
  15. @api.doc(params={'id': 'An ID'})
  16. class MyResource(Resource):
  17.     @api.doc(model=fields)
  18.     def get(self, id):
  19.         return {}
  21.     @api.doc(model='MyModel', body=Person)
  22.     def post(self, id):
  23.         return {}

模型同时也可以被 RequestParser 指定：

  1. parser = api.parser()
  2. parser.add_argument('param', type=int, help='Some param', location='form')
  3. parser.add_argument('in_files', type=FileStorage, location='files')
  5. @api.route('/with-parser/', endpoint='with-parser')
  6. class WithParserResource(restplus.Resource):
  7.     @api.expect(parser)
  8.     def get(self):
  9.         return {}

headers 处理

你可以使用 api.header() 修饰器快捷生成响应头内容。

  1. @api.route('/with-headers/')
  2. @api.header('X-Header', 'Some class header')
  3. class WithHeaderResource(restplus.Resource):
  4.     @api.header('X-Collection', type=[str], collectionType='csv')
  5.     def get(self):
  6.         pass

如果你要指定一个只出现在响应中的标头，只需要使用 @api.response 的 headers 关键字即可:

  1. @api.route('/response-headers/')
  2. class WithHeaderResource(restplus.Resource):
  3.     @api.response(200, 'Success', headers={'X-Header': 'Some header'})
  4.     def get(self):
  5.         pass

对于输入的 headers 进行验证需要使用 @api.expect 修饰符。

  1. parser = api.parser()
  2. parser.add_argument('Some-Header', location='headers')
  4. @api.route('/expect-headers/')
  5. @api.expect(parser)
  6. class ExpectHeaderResource(restplus.Resource):
  7.     def get(self):
  8.         pass

隐藏文档

用下面的方法你可以将一些资源类或函数从文档种隐藏：

  1. # Hide the full resource
  2. @api.route('/resource1/', doc=False)
  3. class Resource1(Resource):
  4.     def get(self):
  5.         return {}
  7. @api.route('/resource2/')
  8. @api.doc(False)
  9. class Resource2(Resource):
  10.     def get(self):
  11.         return {}
  13. @api.route('/resource3/')
  14. @api.hide
  15. class Resource3(Resource):
  16.     def get(self):
  17.         return {}
  19. # Hide methods
  20. @api.route('/resource4/')
  21. @api.doc(delete=False)
  22. class Resource4(Resource):
  23.     def get(self):
  24.         return {}
  26.     @api.doc(False)
  27.     def post(self):
  28.         return {}
  30.     @api.hide
  31.     def put(self):
  32.         return {}
  34.     def delete(self):
  35.         return {}

Swagger 个性化配置

你可以通过 doc 关键字配置 Swagger UI 的路径（默认为根目录）：

  1. from flask import Flask, Blueprint
  2. from flask_restx import Api
  4. app = Flask(__name__)
  5. blueprint = Blueprint('api', __name__, url_prefix='/api')
  6. api = Api(blueprint, doc='/doc/')
  8. app.register_blueprint(blueprint)
  10. assert url_for('api.doc') == '/api/doc/'

你可以通过设定 config.SWAGGER_VALIDATOR_URL 来指定一个自定义的验证器地址（validator URL）：

  1. from flask import Flask
  2. from flask_restx import Api
  4. app = Flask(__name__)
  5. app.config.SWAGGER_VALIDATOR_URL = 'http://domain.com/validator'
  7. api = Api(app)

设置 doc=False 可以完全禁用文档。

  1. from flask import Flask
  2. from flask_restx import Api
  4. app = Flask(__name__)
  5. api = Api(app, doc=False)

项目组织

对于一个复杂的项目而言，接下来，我们将会介绍相关项目组织的最佳实践。

namespace拆分

Flask-RESTX 提供了一种对标 Flask Blueprint 的功能，用于将应用程序分隔为多个 namespace。
例如，如下是一个目录结构的例子：

  1. project/
  2. ├── app.py
  3. ├── core
  4. │   ├── __init__.py
  5. │   ├── utils.py
  6. │   └── ...
  7. └── apis
  8.     ├── __init__.py
  9.     ├── namespace1.py
  10.     ├── namespace2.py
  11.     ├── ...
  12.     └── namespaceX.py
  • app 模块将作为一个遵循经典的Flask模式之一的主程序入口（entry point）。
  • core 模块包含了事务逻辑。实际上你爱叫啥叫啥，甚至可以是很多个包。
  • apis 模块将作为你的API的主入口，你需要在这里导入并注册你的应用程序，而命名空间模块就按照你平时写Flask Blueprint 的时候写就行。

一个namespace的模块要包含模型和资源类的声明：

  1. from flask_restx import Namespace, Resource, fields
  3. api = Namespace('cats', description='Cats related operations')
  5. cat = api.model('Cat', {
  6.     'id': fields.String(required=True, description='The cat identifier'),
  7.     'name': fields.String(required=True, description='The cat name'),
  8. })
  10. CATS = [
  11.     {'id': 'felix', 'name': 'Felix'},
  12. ]
  14. @api.route('/')
  15. class CatList(Resource):
  16.     @api.doc('list_cats')
  17.     @api.marshal_list_with(cat)
  18.     def get(self):
  19.         '''List all cats'''
  20.         return CATS
  22. @api.route('/<id>')
  23. @api.param('id', 'The cat identifier')
  24. @api.response(404, 'Cat not found')
  25. class Cat(Resource):
  26.     @api.doc('get_cat')
  27.     @api.marshal_with(cat)
  28.     def get(self, id):
  29.         '''Fetch a cat given its identifier'''
  30.         for cat in CATS:
  31.             if cat['id'] == id:
  32.                 return cat
  33.         api.abort(404)

apis.init 模块应该用于整合他们：

  1. from flask_restx import Api
  3. from .namespace1 import api as ns1
  4. from .namespace2 import api as ns2
  5. # ...
  6. from .namespaceX import api as nsX
  8. api = Api(
  9.     title='My Title',
  10.     version='1.0',
  11.     description='A description',
  12.     # All API metadatas
  13. )
  15. api.add_namespace(ns1)
  16. api.add_namespace(ns2)
  17. # ...
  18. api.add_namespace(nsX)

此外，你可以在注册API的时候为你的命名空间添加 网址前缀（url-prefixes） 。你不需要在定义命名空间对象的时候指定 网址前缀（url-prefixes） 。

  1. from flask_restx import Api
  3. from .namespace1 import api as ns1
  4. from .namespace2 import api as ns2
  5. # ...
  6. from .namespaceX import api as nsX
  8. api = Api(
  9.     title='My Title',
  10.     version='1.0',
  11.     description='A description',
  12.     # All API metadatas
  13. )
  15. api.add_namespace(ns1, path='/prefix/of/ns1')
  16. api.add_namespace(ns2, path='/prefix/of/ns2')
  17. # ...
  18. api.add_namespace(nsX, path='/prefix/of/nsX')

用这种模式，你只需要在 app.py 中这样注册你的API即可：

  1. from flask import Flask
  2. from apis import api
  4. app = Flask(__name__)
  5. api.init_app(app)
  7. app.run(debug=True)

使用 Blueprints

在 Flask 中提供了 Blueprints 的机制，而在 Flask-RESTX 中，同样保留了对应的机制，以下是一个将 Api 链接到 蓝图（Blueprint） 上的例子：：

  1. from flask import Blueprint
  2. from flask_restx import Api
  4. blueprint = Blueprint('api', __name__)
  5. api = Api(blueprint)

使用蓝图将使你可以将你的API挂载到任何网址前缀（url-prefixes）下，如/或你的应用程序的子域名：

  1. from flask import Flask
  2. from apis import blueprint as api
  4. app = Flask(__name__)
  5. app.register_blueprint(api, url_prefix='/api/1')
  6. app.run(debug=True)

多APIs复用namespace

有时候你可能需要维护一个API的多个版本。如果你使用的时命名空间来搭建你的API，那么将其扩展成多个API是个很简单的事情。
根据先前的布局，我们可以将其迁移到以下目录结构：

  1. project/
  2. ├── app.py
  3. ├── apiv1.py
  4. ├── apiv2.py
  5. └── apis
  6.     ├── __init__.py
  7.     ├── namespace1.py
  8.     ├── namespace2.py
  9.     ├── ...
  10.     └── namespaceX.py

每个 apivX.py 都拥有如下的结构：

  1. from flask import Blueprint
  2. from flask_restx import Api
  4. api = Api(blueprint)
  6. from .apis.namespace1 import api as ns1
  7. from .apis.namespace2 import api as ns2
  8. # ...
  9. from .apis.namespaceX import api as nsX
  11. blueprint = Blueprint('api', __name__, url_prefix='/api/1')
  12. api = Api(blueprint
  13.     title='My Title',
  14.     version='1.0',
  15.     description='A description',
  16.     # All API metadatas
  17. )
  19. api.add_namespace(ns1)
  20. api.add_namespace(ns2)
  21. # ...
  22. api.add_namespace(nsX)

而 app 将这样挂载它们：

  1. from flask import Flask
  2. from api1 import blueprint as api1
  3. from apiX import blueprint as apiX
  5. app = Flask(__name__)
  6. app.register_blueprint(api1)
  7. app.register_blueprint(apiX)
  8. app.run(debug=True)

配置说明

在 Flask-RESTX 中提供了如下一组变量可以用于控制相关的功能行为，如下表所示：

KEY 默认值 功能说明
RESTX_JSON 为 JSON 序列化提供全局配置选项作为 json.dumps() 关键字参数的字典。
RESTX_VALIDATE False 是否在使用 @api.expect() 装饰器时默认强制执行有效参数验证。
RESTX_MASK_HEADER X-Fields MASK功能读取的Headers的key。
RESTX_MASK_SWAGGER True Swagger页面是否启动MASK功能。
RESTX_INCLUDE_ALL_MODELS False 是否在Swagger中包含所有的定义Models，即使它没有被expect或marshal_with装饰器引用。
BUNDLE_ERRORS False 每次校验时是否返回全部错误信息。

完整示例

  1. from flask import Flask
  2. from flask_restx import Api, Resource, fields
  3. from werkzeug.middleware.proxy_fix import ProxyFix
  5. app = Flask(__name__)
  6. app.wsgi_app = ProxyFix(app.wsgi_app)
  7. api = Api(app, version='1.0', title='TodoMVC API',
  8.     description='A simple TodoMVC API',
  9. )
  11. ns = api.namespace('todos', description='TODO operations')
  13. todo = api.model('Todo', {
  14.     'id': fields.Integer(readonly=True, description='The task unique identifier'),
  15.     'task': fields.String(required=True, description='The task details')
  16. })
  19. class TodoDAO(object):
  20.     def __init__(self):
  21.         self.counter = 0
  22.         self.todos = []
  24.     def get(self, id):
  25.         for todo in self.todos:
  26.             if todo['id'] == id:
  27.                 return todo
  28.         api.abort(404, "Todo {} doesn't exist".format(id))
  30.     def create(self, data):
  31.         todo = data
  32.         todo['id'] = self.counter = self.counter + 1
  33.         self.todos.append(todo)
  34.         return todo
  36.     def update(self, id, data):
  37.         todo = self.get(id)
  38.         todo.update(data)
  39.         return todo
  41.     def delete(self, id):
  42.         todo = self.get(id)
  43.         self.todos.remove(todo)
  46. DAO = TodoDAO()
  47. DAO.create({'task': 'Build an API'})
  48. DAO.create({'task': '?????'})
  49. DAO.create({'task': 'profit!'})
  52. @ns.route('/')
  53. class TodoList(Resource):
  54.     '''Shows a list of all todos, and lets you POST to add new tasks'''
  55.     @ns.doc('list_todos')
  56.     @ns.marshal_list_with(todo)
  57.     def get(self):
  58.         '''List all tasks'''
  59.         return DAO.todos
  61.     @ns.doc('create_todo')
  62.     @ns.expect(todo)
  63.     @ns.marshal_with(todo, code=201)
  64.     def post(self):
  65.         '''Create a new task'''
  66.         return DAO.create(api.payload), 201
  69. @ns.route('/<int:id>')
  70. @ns.response(404, 'Todo not found')
  71. @ns.param('id', 'The task identifier')
  72. class Todo(Resource):
  73.     '''Show a single todo item and lets you delete them'''
  74.     @ns.doc('get_todo')
  75.     @ns.marshal_with(todo)
  76.     def get(self, id):
  77.         '''Fetch a given resource'''
  78.         return DAO.get(id)
  80.     @ns.doc('delete_todo')
  81.     @ns.response(204, 'Todo deleted')
  82.     def delete(self, id):
  83.         '''Delete a task given its identifier'''
  84.         DAO.delete(id)
  85.         return '', 204
  87.     @ns.expect(todo)
  88.     @ns.marshal_with(todo)
  89.     def put(self, id):
  90.         '''Update a task given its identifier'''
  91.         return DAO.update(id, api.payload)
  94. if __name__ == '__main__':
  95.     app.run(debug=True)