基于Flask开发企业级REST API应用（三）

阳仔

发布于 2019-08-15 17:26:33

5570

发布于 2019-08-15 17:26:33

关于我编程界的一名小小程序猿，目前在一个创业团队任team lead，技术栈涉及Android、Python、Java和Go，这个也是我们团队的主要技术栈。 Github：https://github.com/hylinux1024 微信公众号：angrycode

前两章把程序的结构以及 API的协议基本上搭建起来了。本文开始不打算对每个模块接口都进行实现，因为基本上都是业务逻辑代码，而且整篇文章都把代码贴出来，那将是一个灾难。

《上一章》对登录授权模块的接口进行了实现，在写本篇文字的时候，我也把用户模块的用户列表、用户信息查询、更新用户信息等接口进行了实现。写到这里的时候我发现，有很多重复的逻辑。比如说，登录参数校验、错误信息处理等这些逻辑，其实这些逻辑可以进行统一处理。

0x00 统一错误处理

客户端如果访问了以下这个没有定义的接口

http://127.0.0.1:5000/api/auth/something

将返回以下信息

Not Found

The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again.

或者有一些数据库操作出错，也会导致服务器的内部错误

Internal Server Error

The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.

这些信息对使用这个系统 API的客户端来说不是很友好，我们希望通过结构化的 json数据进行返回。

要对这种 http协议的错误信息请求统一处理或者实现自定义的错误页面，就需要用到 @errorhandler这个装饰器。

在 app.py中，增加以下两个方法

@app.errorhandler(404)
def not_found_error(error):
    return make_response_error(404, error.description)


@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return make_response_error(500, error.description)

当请求一个不存在的 url时，我们的系统应该返回类似以下的信息

{
"code": 404,
"msg": "The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again."
}

这样就跟我们的定义的数据结构接口协议保持一致。

0x01 统一验证用户token

由于系统中有很多接口是需要用户登录 token才能访问的，所以每个接口都进行登录 token的验证。打开 users.py模块，以下接口都有 token验证的逻辑

@bp.route('/show', endpoint='show')
def show_user_info():
    uid = request.args.get('userId')
    peer_id = request.args.get('peerId')
    token = request.args.get('token', '')
    if not UserInfo.check_token(uid, token):
        return make_response_error(504, 'no operation permission')

    ...
    # 省略不必要的代码
    return make_response_ok(data)
@bp.route('/hot/list', endpoint='list')
def list_hot_user():
    uid = request.args.get('userId')
    token = request.args.get('token', '')
    if not UserInfo.check_token(uid, token):
        return make_response_error(504, 'no operation permission')
    ...
    # 省略不必要的代码
    return make_response_ok(obj)

@bp.route('/update', methods=["POST"], endpoint="update")
def update_user():
    uid = request.form.get('userId', '')
    token = request.form.get('token', '')

    if not UserInfo.check_token(uid, token):
        return make_response_error(504, 'no operation permission')

    ...
    # 省略不必要的代码
    return make_response_ok(data={"data": user.id})

上面三个接口都有相同的验证 token的逻辑

if not UserInfo.check_token(uid, token):
        return make_response_error(504, 'no operation permission')

而这个系统的接口远不止这些，如果每个接口都写相同的逻辑代码，看起来也不怎么优雅。

是不是可以跟前面定义的 @validsign装饰器一样，定义一个 @require_token的装饰器呢？答案是肯定的。

但这里我想直接修改 @validsign这个装饰器函数，给它添加一个参数 @validsign(require_token=True)这种方式，使用起来应该会更加简洁。

def validsign(require_token=False, require_sign=True):
    """
    验证签名,token信息
    :param require_token: 是否验证token
    :param require_sign: 是否验证签名
    :return:
    """
    def decorator(func):
        def wrapper():
            params = _get_request_params()
            if require_sign:
                appkey = params.get('appkey')
                sign = params.get('sign')
                csign = signature(params)
                if not appkey:
                    return make_response_error(300, 'appkey is none.')
                if csign != sign:
                    return make_response_error(500, 'signature is error.')
            if require_token:
                token = params.get('token')
                uid = params.get('userId')
                if not UserInfo.check_token(uid, token):
                    return make_response_error(504, 'no operation permission')
            return func()
        return wrapper
    return decorator

通过参数 require_token和 require_sign可以比较灵活的控制接口的验证逻辑，对开发过程中调试也是很有帮助的。

这里把 token对验证逻辑封装在 UserInfo里面了，这是一个静态方法

@staticmethod
def check_token(uid, token):
    if not token or not uid:
        return False
    user = UserInfo.query.filter_by(id=uid).first()
    if not user:
        return False
    if not user.user_auth:
        return False
    return user.user_auth.token == token

0x02 单元测试

由于对之前的 @validsign装饰器函数进行修改了，单元测试可以验证我们的修改不会影响到具体的业务逻辑，可以保证在原来的基础上进行修改，这是一种保守主义的做事方法。同样地新添加的模块 users.py也需要相应的单元测试功能。

def test_hotlist(self):
    import math
    nonce = math.floor(random.uniform(100000, 1000000))
    params = {'phone': '18922986865', 'userId': '100784', 'appkey': '432ABZ',
              'token': '575f680ddbd0d494a1b5fad8497293d2',
              'timestamp': datetime.now().timestamp(),
              'nonce': nonce}
    sign = signature(params)
    params['sign'] = sign

    respdata = self.app.get("/api/user/hot/list", data=params)

    self.assertEqual(200, respdata.status_code)

    resp = respdata.json
    self.assertEqual(0, resp['code'], respdata.data)
    self.assertIsNotNone(resp['data'], respdata.data)

这个是对首页列表的加载的测试，比较简单。