flask响应的规范化设计

最近使用flask开发一个监控平台的后台，在服务响应上需要考虑到：

服务响应状态码记录，用于排查错误等；
出现错误的请求记录错误的地方，及其错误描述；
可自定义响应信息；
模块化；

于是，又能专心听歌肝活了...

在听了整整两遍歌单的加持下，终于实现自己满意的code。

一、预生成响应类 response.py

from flask import request, g
from apps import logger
from utils import detail_error_log


class NormalResponse(object):
    def __call__(self, code:int=None, msg:str=None, log_msg:str=None):
        code = code if code else self.code
        msg = msg if msg else self.msg

        log_msg_detail = f"{g.user.ljust(22)} {request.path} {request.method} <{code} {self.__class__.__name__}> {g.params} "
        if log_msg:
            if isinstance(log_msg, str):
                log_msg_detail += log_msg
            else:
                log_msg_detail += detail_error_log(log_msg)

        if self.log_level == "INFO":
            logger.info(log_msg_detail)

        elif self.log_level == "ERROR":
            logger.error(log_msg_detail)

        return {"code": code,"msg": msg}

def response(code: int, name: str, msg: str, log_level:str):
    base_cls = (NormalResponse, )
    cls_attr = {'code': code, "msg": msg, "log_level": log_level}
    return type(name, base_cls, cls_attr)()


Ok                              = response(10200, "Ok", "成功", "INFO")

# user 
TokenNotFound                   = response(10401, "TokenNotFound", "未携带token", "INFO")
TokenInvalid                    = response(10402, "TokenInvalid", "无效的token", "INFO")
TokenUpdateFailed               = response(10403, "TokenUpdateError", "token更新错误", "ERROR")
UserNotExists                   = response(10404, "UserNotExists", "用户不存在", "INFO")
LoginFailed                     = response(10405, "LoginError", "登录失败", "ERROR")
UserNotLogin                    = response(10406, "UserNotLogin", "用户未登录", "INFO")
PasswordError                   = response(10407, "PasswordError", "密码错误", "INFO")
UserSignInFailed                = response(10408, "UserSignInFailed", "用户注册失败", "ERROR")
UserExists                      = response(10409, "UserExists", "用户已存在", "INFO")

# client
ParamsMissing                   = response(10410, "ParamsMissing", "缺少参数", "INFO")
MethodNotAllowed                = response(10411, "MethodNotAllowed", "不允许的请求方式", "INFO")
SessionIdError                  = response(10412, "SessionIdError", "任务id错误", "INFO")
GetParamsError                  = response(10413, "GetParamsError", "参数错误", "INFO")
CombinationParamsMissing        = response(10414, "CombinationParamsMissing", "缺少组合参数", "INFO")
SearchKeyNotAvailable           = response(10415, "SearchKeyNotAvailable", "不支持的search_key", "INFO")
IpLimited                       = response(10416, "IpLimit", "IP受限", "INFO")
ParamsError                     = response(10417, "ParamsError", "参数错误", "INFO")

# other
ServerError                     = response(10500, "ServerError", "内部错误", "ERROR")
UnkownError                     = response(10501, "UnkownError", "未知错误", "ERROR")

代码思想主要是利用type函数动态创建类，在类实例化的时候，自定义__call__方法来执行一些具体的操作。

如果在响应类实例化的时候，没有传参，则使用类创建时的默认参数。

NormalResponse基类支持参数：

code 响应状态码
msg 状态码说明或响应数据
log_msg 日志的额外补充

log_msg可传入捕获的异常。

# 提取异常的核心信息
def detail_error_log(err):
    try:
        return f"{err.__traceback__.tb_frame.f_globals['__file__']}[{err.__traceback__.tb_lineno}]: {err}"
    except:
        return ""

二、日志模块

日志模块实现无需特殊配合。自行实现即可，帅气的人可参考本站：Python日志logging封装

三、应用

@api.route('/login', methods=['POST'])
def login():
    '''登录
    TODO: 添加图片验证
    :return 返回响应,保持登录状态
    '''
        
    email = g.params.get("email")
    password = g.params.get("password")
    g.user = email

    if not all([email, password]):
        return ParamsMissing()

    try:
        user:User = User.query.filter(User.email==email).first()

        if not user:
            return UserNotExists()

        elif not user.verify_password(password):
            return PasswordError()

        g.user = user.email
        rftoken = user.generate_user_token(expiration=config.RFTOKEN_EXPIRE)
        token = user.generate_user_token(expiration=config.TOKEN_EXPIRE)
        #更新最后一次登录时间
        user.update_last_seen()
        # mysql 更新 rftoken
        user.update_rftoken(rftoken)
        # redis 更新 token
        user.update_token(token)

        return Ok(msg={"token":token, "rftoken": rftoken})

    except Exception as err:
        return LoginFailed(log_msg=err)

导入定义好的响应类，直接return响应。

完结！

撒花~🌹🌼🌺🌻🌸