官方文档：https://flask-restx.readthedocs.io/en/latest/ https://dev.to/po5i/how-to-set-up-a-rest-api-in-flask-in-5-steps-5b7d

项目结构实践

结合蓝图（Flask蓝图的理解）和命名空间的实践

  1. project/
  2. ├── static/    # 静态文件
  3. ├── templates/    # 模版文件
  4.     ├── index.html    
  5. ├── utils
  6. │   ├── __init__.py
  7. │   ├── app.py    # 创建flask程序实例&链接mongo&加载app.config配置
  8. │   ├── main.py    # 启动flask主程序 
  9. └── apis    # 资源文件（api路由）
  10.     ├── __init__.py    # 添加资源（add_namespace）
  11.     ├── namespace1.py
  1. # apis/namespace1.py
  3. from flask_restx import Resource, fields, Namespace
  4. from flask import render_template, make_response, jsonify
  5. from logzero import logger
  6. from utils.app import DB
  8. # 第一步：将应用程序拆分为可重用的命名空间（组织资源）
  9. api = Namespace(name="test_api", description="这是一个分类描述")
  12. cat = api.model(
  13.     "Cat",
  14.     {
  15.         "id": fields.String(required=True, description="The cat identifier"),
  16.         "name": fields.String(required=True, description="The cat name"),
  17.     },
  18. )
  20. CATS = [
  21.     {"id": "1", "name": "cat1"},
  22. ]
  23. @api.route("/mongo_test")
  24. class MonogoTest(Resource):
  25.     @api.doc("这是一个链接mongo数据库测试接口")
  26.     def get(self):
  27.         """链接mongo数据库"""
  28.         res = DB.test.find_one({})
  29.         logger.info(f"DB:{res}")
  30.         return make_response(jsonify(res['title']))
  33. @api.route("/<id>")
  34. @api.param("id", "The cat identifier")
  35. @api.response(404, "Cat not found")
  36. class Cat(Resource):
  37.     @api.doc("get_cat")
  38.     @api.marshal_with(cat)
  39.     def get(self, id):
  40.         """
  41.         Fetch a cat given its identifier
  42.         args:
  43.             id: 可输入值为1
  44.         """
  45.         for cat in CATS:
  46.             if cat["id"] == id:
  47.                 return cat
  48.         api.abort(404)
  51. """make_respoapie函数在视图内控制响应对象的结果
  52. 如果视图返回的是一个响应对象，那么就直接返回它。
  53. 如果返回的是一个字符串，那么根据这个字符串和缺省参数生成一个用于返回的 响应对象。
  54. 如果返回的是一个字典，那么调用 jsonify 创建一个响应对象。
  55. 如果返回的是一个元组，那么元组中的项目可以提供额外的信息。元组中必须至少 包含一个项目，且项目应当由 (respoapie, status) 、 (respoapie, headers) 或者 (respoapie, status, headers) 组成。 status 的值会重载状态代码， headers 是一个由额外头部值组成的列表 或字典
  56. """
  57. @api.route("/index")
  58. class Home(Resource):
  59.     @api.doc("这是一个打开index.html测试接口")
  60.     def get(self):
  61.         """打开index.html"""
  62.         resp = make_response(render_template("index.html"), 404)
  63.         resp.headers["x-Something"] = "S value"
  64.         return resp
  67. todo = api.model('Todo', {
  68.     'id': fields.Integer(readonly=True, description='The task unique identifier'),
  69.     'task': fields.String(required=True, description='The task details')
  70. })
  73. class TodoDAO(object):
  74.     def __init__(self):
  75.         self.counter = 0
  76.         self.todos = []
  78.     def get(self, id):
  79.         for todo in self.todos:
  80.             if todo['id'] == id:
  81.                 return todo
  82.         api.abort(404, "Todo {} doesn't exist".format(id))
  84.     def create(self, data):
  85.         todo = data
  86.         todo['id'] = self.counter = self.counter + 1
  87.         self.todos.append(todo)
  88.         return todo
  90.     def update(self, id, data):
  91.         todo = self.get(id)
  92.         todo.update(data)
  93.         return todo
  95.     def delete(self, id):
  96.         todo = self.get(id)
  97.         self.todos.remove(todo)
  100. DAO = TodoDAO()
  102. @api.route('/')
  103. class TodoList(Resource):
  104.     '''Shows a list of all todos, and lets you POST to add new tasks'''
  105.     @api.doc('list_todos')
  106.     @api.marshal_list_with(todo)
  107.     def get(self):
  108.         '''List all tasks'''
  109.         return DAO.todos
  111.     @api.doc('create_todo')
  112.     @api.expect(todo)
  113.     @api.marshal_with(todo, code=201)
  114.     def post(self):
  115.         '''Create a new task'''
  116.         return DAO.create(api.payload), 201
  119. @api.route('/<int:id>')
  120. @api.response(404, 'Todo not found')
  121. @api.param('id', 'The task identifier')
  122. class Todo(Resource):
  123.     '''Show a single todo item and lets you delete them'''
  124.     @api.doc('get_todo')
  125.     @api.marshal_with(todo)
  126.     def get(self, id):
  127.         '''Fetch a given resource'''
  128.         return DAO.get(id)
  130.     @api.doc('delete_todo')
  131.     @api.response(204, 'Todo deleted')
  132.     def delete(self, id):
  133.         '''Delete a task given its identifier'''
  134.         DAO.delete(id)
  135.         return '', 204
  137.     @api.expect(todo)
  138.     @api.marshal_with(todo)
  139.     def put(self, id):
  140.         '''Update a task given its identifier'''
  141.         return DAO.update(id, api.payload)
  144.   ---------------------------------------------------------------------------------------------------------------------------
  146. # apis/__init__.py
  148. from flask_restx import Api
  149. from .flask_restx_test import Todo, api as resrx_test_api
  150. from flask import Blueprint
  151. from .flask_restx_test import Home, Cat, MonogoTest, TodoList
  154. # 第二步: 创建蓝图
  155. blueprint = Blueprint("api", __name__, url_prefix="/api/v1")
  157. # 第三步：初始化应用程序入口
  158. api = Api(blueprint, title="Zaygee API", version="1.0", description="A simple demo API",)
  161. # Todo: 这种方式添加资源在api文档中无法正确展示命名空间的分类
  162. # 第五步：为给定的 API 命名空间注册资源
  163. # api.add_resource(Home, "/index")
  164. # api.add_resource(MonogoTest, "/mongo_test")
  165. # api.add_resource(Cat, "/<id>")
  166. # api.add_resource(TodoList, "/list")
  169. # 第四步：为api的当前实例注册命名空间
  170. api.add_namespace(resrx_test_api)
  173. ---------------------------------------------------------------------------------------------------------------------------
  174. # utils/app.py
  176. from logzero import logger
  177. from flask_pymongo import PyMongo
  178. from flask import Flask
  179. from dataclasses import dataclass   # 定义数据类
  180. from werkzeug.middleware.proxy_fix import ProxyFix
  181. from pathlib import Path
  183. # 静态文件&模板文件目录
  184. static_folder = ''.join([str(Path(__file__).parent.parent), "/static"])
  185. template_folder = ''.join([str(Path(__file__).parent.parent), "/templates"])
  188. """创建flask程序实例&链接mongo"""
  190. app = Flask(__name__, static_folder=static_folder, template_folder=template_folder)
  191. app.wsgi_app = ProxyFix(app.wsgi_app)
  193. # 链接mongodb
  194. app.config["MONGO_URI"] = "mongodb://localhost:27017/test?authSource=admin"    # 本地mongo服务
  195. mongo = PyMongo(app)
  196. DB = mongo.db
  197. logger.info(f"成功链接mongo: {DB}")
  200. ---------------------------------------------------------------------------------------------------------------------------
  202. # utils/main.py
  204. from flask_demo.utils.app import app
  205. from api_resource import blueprint as api1
  208. # 第六步：延迟注册app
  209. # api.init_app(app)
  210. # 第六步：注册蓝图
  211. app.register_blueprint(api1)
  213. # 第七步：执行应用程序
  214. app.run(debug=True)

路由&端点

  1. from flask_restx import Resource, Namespace
  2. from flask import make_response, jsonify, url_for
  3. from logzero import logger
  4. from utils.app import DB
  6. api = Namespace(name="route", description="接口路由&端点")
  9. @api.route("/hello", "/world", endpoint='mutil_endpoint')
  10. class MonogoTest(Resource):
  11.     @api.doc("这是一个多url的")
  12.     def get(self):
  13.         """多url接口"""
  14.         return "hello world!"

image.png

请求参数解析

location参数可指定提取参数的位置，可选如下：

args：路径参数 form：表单参数 headers：请求头参数 json：json请求体 values：请求体或查询参数

bundle_errors参数设置=True，可绑定错误处理一次返回

parser = reqparse.RequestParser(bundle_errors=True)

解析器继承

  1. from flask_restx import reqparse
  3. parser = reqparse.RequestParser()
  4. parser.add_argument('foo', type=int)
  6. # 复制
  7. parser_copy = parser.copy()
  8. parser_copy.add_argument('bar', type=int)
  10. # 替换
  11. parser_copy.replace_argument('foo', required=True, location='json')
  13. # 删除某个参数
  14. parser_copy.remove_argument('foo')
  1. # demo
  3. from flask_restx import reqparse, Resource, Namespace, abort
  4. from logzero import logger
  7. api = Namespace(name="reparse_test", description="请求参数解析")
  9. todos = {"a": "111", "2": "222"}
  11. @api.route("/todo")
  12. class Todo(Resource):
  14.     # 请求参数解析，增加参数验证
  15.     parser = reqparse.RequestParser()
  16.     parser.add_argument('data', type=str, help='请求参数必须是字符串', required=True, location="json")
  17.     parser.add_argument('todo_id', type=str, help='请求参数必须是字符串', required=True, location="json")
  18.     parser.add_argument('age', type=int, help='请求参数必须是int', required=True, location="json")
  20.     req_parser = reqparse.RequestParser()
  21.     req_parser.add_argument('todo_id', type=str, help='请求参数必须是字符串', required=True, location="args")
  22.     req_parser.add_argument('age', type=int, help='请求参数必须是int', required=True, location="args")
  25.     @api.expect(parser)
  26.     def put(self):
  27.         """put请求-更新数据"""
  28.         # strict=True 为强校验，包含解析器未定义的参数则抛出异常
  29.         args = self.parser.parse_args(strict=True)
  30.         todo_id = args["todo_id"]
  31.         data = args["data"]
  32.         age = args["age"]
  33.         todos[todo_id] = {"data": data, "age": age}
  34.         return {todo_id: todos[todo_id]}, 201
  36.     @api.expect(parser)
  37.     def post(self):
  38.         """post请求-创建数据"""
  39.         args = self.parser.parse_args(strict=True)
  40.         todo_id = args["todo_id"]
  41.         data = args["data"]
  42.         age = args["age"]
  43.         todos[todo_id] = {"data": data, "age": age}
  44.         return {
  45.             "status_code": 200,
  46.             "msg": "success"
  47.         }
  49.     @api.expect(req_parser)
  50.     def get(self):
  51.         """get请求-获取数据"""
  52.         args = self.req_parser.parse_args(strict=True)
  53.         todo_id = args["todo_id"]
  54.         logger.info(f'todo_id: {todo_id}')
  55.         age = args['age']
  56.         logger.info(f'age: {age}')
  57.         if todos.get(todo_id) and dict(todos.get(todo_id)).get("age") == age:
  58.             return {todo_id: todos.get(todo_id)}
  59.         else:
  60.             abort(404, message="Todo {} doesn't exist".format(todo_id))
  63.     @api.expect(req_parser)
  64.     def delete(self):
  65.         """delete请求-删除数据"""
  66.         args = self.req_parser.parse_args(strict=True)
  67.         todo_id = args["todo_id"]
  68.         age = args['age']
  70.         if todos.get(todo_id) and dict(todos.get(todo_id)).get("age") == age:
  71.             del todos[todo_id]
  72.             return {
  73.             "status_code": 204,
  74.             "msg": "success"
  75.         }
  76.         else:
  77.             abort(404, message="Todo {} 删除失败".format(todo_id))

image.png

响应结果验证

flask-restx通过fields模块提供控制在响应中呈现的数据或者期望作为输入的数据

Swagger文档

  1. @api.route():装饰允许你设置接口路由
  2. @api.doc():装饰允许你包含在文档中的附加信息
  3. @api.expect():装饰允许您指定预期的输入字段
  5. # demo
  7. from flask import Flask
  8. from flask_restx import Api, Resource, fields
  10. app = Flask(__name__)
  11. api = Api(app, version='1.0', title='Sample API',
  12.     description='A sample API',
  13. )
  15. @api.route('/my-resource/<id>')
  16. @api.doc(params={'id': 'An ID'})
  17. class MyResource(Resource):
  18.     def get(self, id):
  19.         return {}
  21.     @api.response(403, 'Not Authorized')
  22.     def post(self, id):
  23.         api.abort(403)
  26. if __name__ == '__main__':
  27.     app.run(debug=True)