支持Flask快速构建REST API 官方文档：https://flask-restful.readthedocs.io/en/latest/

快速安装
  1. pip3 install flask-restful

使用示例
  1. from flask import Flask, request, Response
  2. from flask_restful import Resource, Api, abort
  4. app = Flask(__name__)
  5. api = Api(app)
  7. todos = {}
  9. """继承资源类Resource，在类中定义方法"""
  10. class TodoSimple(Resource):
  11.     """restful 接口"""
  12.     def get(self, todo_id):
  13.         """get请求-获取数据"""
  14.         if not todos.get(todo_id):
  15.             # abort(Response("查询失败", 404))
  16.             abort(404, message="Todo {} doesn't exist".format(todo_id))
  17.         return {todo_id: todos.get(todo_id)}
  19.     def put(self, todo_id):
  20.         """put请求-更新数据"""
  21.         todos[todo_id] = request.form.get("data")
  22.         return {todo_id: todos[todo_id]}
  24.     def post(self, todo_id):
  25.         """post请求-创建数据"""
  26.         todos[todo_id] = request.form.get("data")
  27.         return {
  28.             "status_code": 200,
  29.             "msg": "success"
  30.         }
  32.     def delete(self, todo_id):
  33.         """delete请求-删除数据"""
  34.         todos[todo_id] = request.form.get("data")
  35.         if not todos.get(todo_id):
  36.             abort(Response('删除失败', 404))
  37.         del todos[todo_id]
  38.         return {
  39.             "status_code": 200,
  40.             "msg": "success"
  41.         }
  44. """向 api 添加资源"""
  45. api.add_resource(TodoSimple, '/<string:todo_id>')
  46. # 设置端点
  47. # api.add_resource(TodoSimple, '/<string:todo_id>', endpoint="home")
  50. if __name__ == "__main__":
  51.     app.run(debug=True)

参数解析-reqparse
  1. parser = reqparse.RequestParser(bundle_errors=True)
  2. parser.add_argument('data', type=str, help='请求参数必须是字符串', required=True, location='form')
  3. args = self.parser.parse_args(strict=True)

通过flask_restful内置的reqparse，提供对请求数据验证的支持

  1. strict=True： 为强校验，包含解析器未定义的参数则抛出异常
  2. required=True： 可设置参数为必填参数
  3. location：可指定提取值的位置：form、args、headers、cookies、files
  4. bundle_errors=True：将错误捆绑在一起并一次发送回客户端

更多见：https://flask-restful.readthedocs.io/en/latest/reqparse.html#basic-arguments

  1. # 使用实例
  3. from flask_restful import Resource, Api, abort, reqparse
  5. class TodoSimple(Resource):
  6.     """restful 接口"""
  8.     # 请求参数解析，增加参数验证
  9.     parser = reqparse.RequestParser()
  10.     parser.add_argument('data', type=str, help='请求参数必须是字符串')
  12.     def put(self, todo_id):
  13.         """put请求-更新数据"""
  14.         # strict=True 为强校验，包含解析器未定义的参数则抛出异常
  15.         args = self.parser.parse_args(strict=True)
  16.         # todos[todo_id] = request.form.get("data")
  17.         todos[todo_id] = args["data"]
  18.         return {todo_id: todos[todo_id]}, 201
  20.     def post(self, todo_id):
  21.         """post请求-创建数据"""
  22.         args = self.parser.parse_args(strict=True)
  23.         todos[todo_id] = args["data"]
  24.         # todos[todo_id] = request.form.get("data")
  25.         return {
  26.             "status_code": 200,
  27.             "msg": "success"
  28.         }

输出字段-Fields

控制响应数据的呈现 1.可以在资源中使用任何您想要的对象（ORM 模型/自定义类/等） 2.还可以让您格式化和过滤响应，以防暴露内部数据结构 envelope:指定可选关键字参数包装结果输出 marshal_with装饰器：获取您的数据对象并应用字段过滤 更多见：https://flask-restful.readthedocs.io/en/latest/fields.html

  1. from flask import Flask
  2. from flask_restful import Resource, Api, abort, reqparse, fields, marshal_with
  3. from typing import Optional
  5. app = Flask(__name__)
  6. api = Api(app)
  8. resource_fields = {
  9.     'task': fields.String,
  10.     "todo_id": fields.String,
  11.     "status": fields.String(default="success")
  12.     }
  14. class TodoDao(object):
  15.     """响应参数类"""
  16.     def __init__(self, todo_id, task, status: Optional[str] = None):
  17.         self.todo_id = todo_id
  18.         self.task = task
  19.         self.status = status
  22. class Todo(Resource):
  23.     """数据格式化"""
  24.     # envelope指定可选关键字参数包装结果输出
  25.     # marshal_with装饰器：获取您的数据对象并应用字段过滤
  26.     # @marshal_with(resource_fields)
  27.     @marshal_with(resource_fields, envelope='resource')
  28.     def get(self, **kwargs):
  29.         """get"""
  30.         return TodoDao(todo_id="todo2", task='Remember the milk')
  33. """向 api 添加资源"""
  34. api.add_resource(Todo, "/todo_resource")
  37. if __name__ == "__main__":
  38.     app.run(debug=True)
  41. """
  42. # 响应示例
  43. {
  44.     "resource": {
  45.         "task": "Remember the milk",
  46.         "todo_id": "todo2",
  47.         "status": "success"
  48.     }
  49. }
  50. """

项目结构实践
  1. /
  2.     /api_resource/    :存放接口资源
  3.         /__init.py
  4.     /common/          ：存放公共配置
  5.         /config.py
  6.         /__init.py
  7.     /static/          ：静态文件
  8.     /templates/       ：模版
  9.     /utils/           
  10.         /__init.py
  11.         /app.py       ：flask程序实例&链接mongo
  12.         /main.py      ：向api添加资源，启动flask主程序
  1. # /utils/app.py
  3. from flask_restful import Api
  4. from logzero import logger
  5. from flask_pymongo import PyMongo
  6. from flask import Flask
  7. from common import config
  9. """创建flask程序实例&链接mongo"""
  10. app = Flask(__name__)
  11. api = Api(app)
  13. """链接mongodb"""
  14. app.config["MONGO_URI"] = ''.join([config.mongodb_str, "/", config.MONGODB_DBNAME, "?authSource=admin"])
  15. mongo = PyMongo(app)
  16. DB = mongo.db
  17. logger.info(f"成功链接mongo: {DB}")
  21. # /api_resource/api_test.py
  23. from flask import jsonify
  24. from flask_restful import Resource, abort
  25. from logzero import logger
  26. from utils.app import DB
  29. class TestApi(Resource):
  30.     """restful api"""
  31.     def get(self):
  32.         """get"""
  33.         logger.info("restful test")
  34.         logger.info(DB.test.find_one({}))
  35.         return jsonify(message="success")
  38. # /utils/main.py
  40. """向api添加资源，启动flask主程序"""
  41. from api_resource.api_test import TestApi
  42. from logzero import logger
  43. from app import api, app
  46. api.add_resource(TestApi, "/add_one")
  49. if __name__ == "__main__":
  50.     app.run(host="0.0.0.0", debug=True)