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

关于我
编程界的一名小小程序猿，目前在一个创业团队任team lead，技术栈涉及Android、Python、Java和Go，这个也是咱们团队的主要技术栈。
Github：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)
复制代码

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

0x03 小结一下

在项目开发过程当中，对于重复的逻辑应该要抽象封装

Don't repeat yourself 复制代码

而如何封装就要看我的功力了，我以为除了多学习，多看源码，几乎没有其它捷径。

0x04 学习资料

• palletsprojects.com/p/flask/
  flask官方文档
• docs.sqlalchemy.org/en/13/orm/b…
  models关系映射相关文档
• github.com/hylinux1024… 本文项目源码