Flask - Restful API 开发

发表于 更新于 分类于 > > 本文字数： 10k 阅读时长 ≈ 17 分钟

什么是 Restful?

  • REST 即表述性状态传递（英文：Representational State Transfer，简称 REST）是 Roy Fielding 博士在 2000 年他的博士论文中提出来的一种软件架构风格。它是一种针对网络应用的设计和开发方式，可以降低开发的复杂性，提高系统的可伸缩性。
  • Restfull 一种软件架构风格、设计风格，而不是标准，只是提供了一组设计原则和约束条件。它主要用于客户端和服务器交互类的软件。基于这个风格设计的软件可以更简洁，更有层次，更易于实现缓存等机制。
  • 满足 rest 这种架构风格的 API 就是 Restful API
  • Restful 接口开发都是围绕资源以及对资源的各种操作展开的

什么是资源？

  • 所谓资源就是在网络上存在的任意实体，哪怕是一条信息。

各种操作？

数据格式

  • 通常数据的传输都采用 json 格式，有时也会采用 GET

调试工具

  • postman 就是一个非常好用的测试工具，可以轻松模拟 HTTP 各种请求
  • 安装使用，一路 next 即可完成安装，傻瓜式操作

原生实现

  • 获取所有资源

    1
    2
    3
    4
    # 获取所有资源
    @app.route('/posts/')
    def get_posts_list():
        return jsonify({'posts': posts})

  • 获取指定资源

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    # 获取指定资源
    @app.route('/posts/<int:pid>')
    def get_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        return jsonify({'posts': p[0]})


    # 定制404错误
    @app.errorhandler(404)
    def page_not_found(e):
        return jsonify({'Error': 'Page Not Found'}), 404

  • 创建新的资源

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    # 创建新的资源
    @app.route('/posts/', methods=['POST'])
    def create_posts():
        if not request.json or 'title' not in request.json or 'content' not in request.json:
            abort(400)
        # 创建新的资源
        p = {
            'id': posts[-1]['id'] + 1,
            'title': request.json.get('title'),
            'content': request.json['content']
        }
        # 保存资源
        posts.append(p)
        return jsonify({'posts': p}), 201


    # 定制400错误
    @app.errorhandler(400)
    def bad_request(e):
        return jsonify({'Error': 'Bad Request'}), 400
    • 指定传输数据类型为 json，Content-Type = application/json
    • 准备 JSON 数据，body 类型选择为 raw，json 数据只能使用双引号，最后的逗号不要加

  • 修改指定资源

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    # 修改指定资源
    @app.route('/posts/<int:pid>', methods=['PUT'])
    def update_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        if 'title' in request.json:
            p[0]['title'] = request.json.get('title')
        if 'content' in request.json:
            p[0]['content'] = request.json['content']
        return jsonify({'posts': p[0]})


    # 定制404错误
    @app.errorhandler(404)
    def page_not_found(e):
        return jsonify({'Error': 'Page Not Found'}), 404

  • 删除指定资源

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    # 删除指定的资源
    @app.route('/posts/<int:pid>', methods=['DELETE'])
    def delete_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        posts.remove(p[0])
        return jsonify({'Result': '数据已删除'})


    # 定制404错误
    @app.errorhandler(404)
    def page_not_found(e):
        return jsonify({'Error': 'Page Not Found'}), 404

  • 完整代码示例

    1
    2
    # settings.py
    ENV = 'development'
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    # app.py
    from flask import Flask, jsonify, abort, request
    from flask_script import Manager

    import settings

    app = Flask(__name__)
    app.config.from_object(settings)
    manager = Manager(app)

    # 测试数据
    posts = [
        {
            'id': 1,
            'title': 'Python基础',
            'content': '别人都说Python语法很简单，但是每次问题都出在语法上'
        },
        {
            'id': 2,
            'title': 'Web前端',
            'content': '不就是几个标签的事嘛， 但是最好细心点'
        }
    ]


    # 获取所有资源
    @app.route('/posts/')
    def get_posts_list():
        return jsonify({'posts': posts})


    # 获取指定资源
    @app.route('/posts/<int:pid>')
    def get_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        return jsonify({'posts': p[0]})


    # 创建新的资源
    @app.route('/posts/', methods=['POST'])
    def create_posts():
        if not request.json or 'title' not in request.json or 'content' not in request.json:
            abort(400)
        # 创建新的资源
        p = {
            'id': posts[-1]['id'] + 1,
            'title': request.json.get('title'),
            'content': request.json['content']
        }
        # 保存资源
        posts.append(p)
        return jsonify({'posts': p}), 201


    # 修改指定资源
    @app.route('/posts/<int:pid>', methods=['PUT'])
    def update_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        if 'title' in request.json:
            p[0]['title'] = request.json.get('title')
        if 'content' in request.json:
            p[0]['content'] = request.json['content']
        return jsonify({'posts': p[0]})


    # 删除指定的资源
    @app.route('/posts/<int:pid>', methods=['DELETE'])
    def delete_posts(pid):
        p = list(filter(lambda p: p['id'] == pid, posts))
        if not p:
            abort(404)
        posts.remove(p[0])
        return jsonify({'Result': '数据已删除'})


    # 定制400错误
    @app.errorhandler(400)
    def bad_request(e):
        return jsonify({'Error': 'Bad Request'}), 400


    # 定制404错误
    @app.errorhandler(404)
    def page_not_found(e):
        return jsonify({'Error': 'Page Not Found'}), 404


    if __name__ == '__main__':
        manager.run()

Flask-RESTful

  • 说明：实现了 RESTful API 开发的拓展库
  • 安装：pip install Flask-RESTful
  • 使用：
    • 导入类库，创建 Api 对象
    • 定义资源类，要继承自 Resource
    • 将资源添加到 api 对象中
    • 资源类中只需要写与请求方法同名的成员方法即可，系统会自动根据请求方法调用对应的函数
  • 示例：
    1
    2
    # settings.py
    ENV = 'development'
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    # app.py
    from flask import Flask
    from flask_script import Manager
    from flask_restful import Resource, Api

    import settings

    app = Flask(__name__)
    app.config.from_object(settings)
    # api = Api(app)
    api = Api()
    manager = Manager(app)


    # 创建资源类
    class UserAPI(Resource):
        def get(self, uid):
            return {'USER': 'GET'}

        def put(self, uid):
            return {'USER': 'PUT'}

        def delete(self, uid):
            return {'USER': 'DELETE'}


    # 一个完整的资源通常需要两个资源类
    class UserListAPI(Resource):
        def get(self):
            return {'UserList': 'GET'}

        def post(self):
            return {'UserList': 'POST'}


    @app.route('/')
    def hello_world():  # put application's code here
        return 'Hello World!'


    # 将资源添加到api中，可以指定多个路由
    api.add_resource(UserAPI, '/users/<int:uid>', '/u/<int:uid>')
    api.add_resource(UserListAPI, '/users/')
    # 若创建Api对象时没有指定app参数，那么后来初始化操作一定要放在添加资源之后
    api.init_app(app)

    if __name__ == '__main__':
        manager.run()

身份认证

  • 说明：Flask-HTTPAuth 提供了基本的身份认证
  • 安装：pip install Flask-HTTPAuth
  • 使用：
    • 导入类库，创建认证对象
    • 书写认证回调函数
    • 在需要认证的路由上添加装饰器 @auth.login_required
    • 在 Flask-HTTPAuth 中，如果一个资源类的所有方法都需要认证，需要将装饰器函数写在 decorators 指定的列表中
  • 示例：
    • 原生身份认证 
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      from flask_httpauth import HTTPBasicAuth

      auth = HTTPBasicAuth()


      # 认证回调函数，返回True表示认证成功，False表示失败
      @auth.verify_password
      def verify_password(username, password):
          if username == 'admin' and password == '123456':
              return True
          return False


      # 获取所有资源
      @app.route('/posts/')
      # 路由保护，需要认证成功才可访问
      # 自动调用@auth.verify_password装饰器修饰的函数
      @auth.login_required
      def get_posts_list():
          return jsonify({'posts': posts})


      # 认证失败错误显示，认证失败时会自动调用
      @auth.error_handler
      def unauthorized_access():
          return jsonify({'Error': 'Unauthorized Access'}), 401
    • Flask-RESTful 身份认证 
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      from flask_restful import Resource, Api
      from flask_httpauth import HTTPBasicAuth

      api = Api(app)
      auth = HTTPBasicAuth()


      # 认证回调函数，返回True表示认证成功，False表示失败
      @auth.verify_password
      def verify_password(username, password):
          if username == 'admin' and password == '123456':
              return True
          return False


      # 认证失败错误显示，认证失败时会自动调用
      @auth.error_handler
      def unauthorized_access():
          return jsonify({'Error': 'Unauthorized Access'}), 401

      # 创建资源类
      class UserAPI(Resource):
          # 添加认证，将需要的装饰器函数写在列表中即可
          decorators = [auth.login_required]

          def get(self, uid):
              return {'USER': 'GET'}

          def put(self, uid):
              return {'USER': 'PUT'}

          def delete(self, uid):
              return {'USER': 'DELETE'}


      # 一个完整的资源通常需要两个资源类
      class UserListAPI(Resource):
          # 路由保护，需要认证成功才可访问
      	# 自动调用@auth.verify_password装饰器修饰的函数
          @auth.login_required
          def get(self):
              return {'UserList': 'GET'}

          def post(self):
              return {'UserList': 'POST'}

基于 token 的身份认证

  • 先写一个生成 token 的路由
  • 用户需要先带着身份信息获取 token
  • 以后再访问需要认证的资源时只需要带着 token 即可
  • 示例 
    1
    2
    3
    # settings.py
    ENV = 'development'
    SECRET_KEY = '123456'
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    # app.py
    from flask import Flask, jsonify, g
    from flask_script import Manager
    from flask_restful import Resource, Api
    from flask_httpauth import HTTPBasicAuth
    from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

    import settings

    app = Flask(__name__)
    app.config.from_object(settings)
    # api = Api(app)
    api = Api()
    auth = HTTPBasicAuth()
    manager = Manager(app)


    # 生成token
    @app.route('/get_token/')
    @auth.login_required
    def generate_token():
        s = Serializer(app.config.get('SECRET_KEY'), expires_in=3600)
        return s.dumps({'username': g.username})


    # 认证回调函数，返回True表示认证成功，False表示失败
    @auth.verify_password
    def verify_password(username_or_token, password):
        if username_or_token == 'admin' and password == '123456':
            g.username = username_or_token
            return True
        # 再次尝试token认证
        s = Serializer(app.config.get('SECRET_KEY'))
        try:
            data = s.loads(username_or_token)
            g.username = data['username']
            return True
        except:
            return False


    # 认证失败错误显示
    @auth.error_handler
    def unauthorized_access():
        return jsonify({'Error': 'Unauthorized Access'}), 401


    # 创建资源类
    class UserAPI(Resource):
        # 添加认证，将需要的装饰器函数写在列表中即可
        decorators = [auth.login_required]

        def get(self, uid):
            return {'USER': 'GET'}

        def put(self, uid):
            return {'USER': 'PUT'}

        def delete(self, uid):
            return {'USER': 'DELETE'}


    # 一个完整的资源通常需要两个资源类
    class UserListAPI(Resource):
        @auth.login_required
        def get(self):
            return {'UserList': 'GET'}

        def post(self):
            return {'UserList': 'POST'}


    @app.route('/')
    def hello_world():  # put application's code here
        return 'Hello World!'


    # 将资源添加到api中，可以指定多个路由
    api.add_resource(UserAPI, '/users/<int:uid>', '/u/<int:uid>')
    api.add_resource(UserListAPI, '/users/')
    # 若创建Api对象时没有指定app参数，那么后来初始化操作一定要放在添加资源之后
    api.init_app(app)

    if __name__ == '__main__':
        manager.run()