Restful API 开发

什么是Restful?

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

什么是资源?

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

各种操作?

  • 所谓操作就是资源的CURD。在开发者设计良好的前期下,对网络的操作都可以抽象为对资源的CURD
  • Restful对资源操作抽象为HTTP协议的GETPOSTDELETE等请求,以完成对资源的CURD具体对照。
    方法 行为 示例
    GET 获取所有资源 http://127.0.0.1/source
    GET 获取指定资源 http://127.0.0.1/source/250
    POST 创建新的资源 http://127.0.0.1/source
    PUT 更新指定资源 http://127.0.0.1/source/250
    DELETE 删除指定资源 http://127.0.0.1/source/250

数据格式

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

调试工具

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

原生实现

  • 获取所有资源
  • 获取指定资源
  • 创建新的资源
    • 指定出传输数据类型为jsonContent-Type = application/json
    • 准备JSON数据,body类型选择为rawjson数据只能使用双引号,最后的逗号不要加
  • 修改指定资源
  • 删除指定资源

flask-restful

  • 说明:实现了restful api开发的拓展库
  • 安装:pip install flask-restful
  • 使用:
    • 导入类库,创建Api对象
    • 定义资源类,要继承自Resource
    • 将资源添加到api对象中
    • 资源类中只需要写与请求方法同名的成员方法即可,系统会自动根据请求方法调用对应的函数

身份认证

  • 说明:flask-httpauth提供了基本的身份认证
  • 安装:pip install flask-httpauth
  • 使用:
    • 导入类库,创建认证对象
    • 书写认证回调函数
    • 在需要认证的路由上添加装饰器@auth.login_required
    • flask-restful中,如果一个资源类的所有方法都需要认证,需要将装饰器函数写在decoration指定的列表中

基于token的身份认证

  • 先写一个生成token的路由
  • 用户需要先带着身份信息获取token
  • 以后再访问需要认证的资源时只需要带着token即可

示例

  • 示例1:

    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
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    from flask import Flask, jsonify, abort, request
    from flask_script import Manager
    from flask_httpauth import HTTPBasicAuth

    auth = HTTPBasicAuth()


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


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


    app = Flask(__name__)
    manager = Manager(app)

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


    # 获取所有资源
    @app.route('/posts/')
    # 路由保护,需要认证成功才可访问
    @auth.login_required
    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 len(p) == 0:
    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['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 len(p) == 0:
    abort(404)
    if 'title' in request.json:
    p[0]['title'] = request.json['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 len(p) == 0:
    abort(404)
    posts.remove(p[0])
    return jsonify({'result': '数据已删除'})


    @app.route('/')
    def index():
    return 'Rest API原生实现'


    @app.errorhandler(404)
    def page_not_found(e):
    return jsonify({'error': 'page not found'}), 404


    @app.errorhandler(400)
    def bad_request(e):
    return jsonify({'error': 'bad request'}), 400


    if __name__ == '__main__':
    manager.run()
  • 示例2:

    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
    from flask import Flask, jsonify, g
    from flask_script import Manager
    from flask_restful import Api, Resource
    from flask_httpauth import HTTPBasicAuth
    from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

    auth = HTTPBasicAuth()
    app = Flask(__name__)

    app.config['SECRET_KEY'] = '123456'
    manager = Manager(app)

    api = Api()


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


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


    # 认证失败错误显示
    @auth.error_handler
    def unauthorized():
    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'}


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


    @app.route('/')
    def index():
    return 'flask-restful'


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