ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > Flask + Ant Vue 实战 （5）-统一API返回json和全局异常处理

Flask + Ant Vue 实战 （5）-统一API返回json和全局异常处理

原创 数据库开发技术 作者：程序员小钟同学 时间：2021-02-24 17:07:12 0 删除 编辑

前言

在做相关的API接口的开发的时候，避免不了，就是我们的需要统一一个返回JSON格式.

前端和后端进行交互，我们一般要求前端按照约定请求URL路径，并传入相关参数，flask后端服务器接收请求，进行业务处理之后返回数据给前端.而后端按约定好的格式返回也是有必要。如我们的实战项目中返回的格式一样：

{
  "code": 0, 
  "message": "\u64cd\u4f5c\u6210\u529f\uff01", 
  "result": [
    {
      "ip": 2, 
      "tian": "2020-12-21", 
      "type": "12-21", 
      "visit": 5
    }, 
    {
      "ip": 1, 
      "tian": "2020-12-22", 
      "type": "12-22", 
      "visit": 5
    }, 
    {
      "ip": 1, 
      "tian": "2020-12-24", 
      "type": "12-24", 
      "visit": 3
    }, 
    {
      "ip": 1, 
      "tian": "2020-12-25", 
      "type": "12-25", 
      "visit": 2
    }
  ], 
  "success": true, 
  "timestamp": 1608875811072
}

其中：

{
 code:int,   #返回状态码
 message:str, #返回信息描述
 result:object,  #返回结果内容信息
 success：bool, # 请求是否处理成功
 timestamp：str, # 请求响应时间戳
}

规范统一的Code状态码

规范统一的Code码通常可以让我们的很快的定位到对应的错误信息。但是我们的通常需要分开和的HTTP状态码区别开。 通常课件的HTTP状态码如下：

分类 区间 分类说明描述
1** 100~199 信息，后端收到请求，需要客户端继续执行操作
2** 200~299 请求响应成功，后端也处理完成，前端也接收数据成功
3** 300~399 重定向，请求链接被重定向都另一个地址，需要继续再执行请求
4** 400~499 客户端错误，请求异常，无法完成请求处理
5** 500~599 后端服务异常，或后端处理请求超时，后端服务无法完成请求（如：502，已下线）

自定义我们的Code码，可以针对性进行设计

如腾讯的返回：

这样开发人员可以在得到返回值后，根据状态码就查阅相关文档，以及message相关的信息描述，可以快速定位错误问题。

flask下的统一API返回格式

API返回就是响应体报文内容信息的定义，Flask中其实已经定义好一个函数jsonify，它可以返回我们的定义好的字典，并起订转化JSON格式响应。

如我们的默认注册的路由中：

def __health_chcek_route(app):
    pass

    @app.route('/backend_health')
    def backend_health():
        return 'ok'

    @app.route('/')
    def route_map():
        """
        主视图，返回所有视图网址
        """
        rules_iterator = app.url_map.iter_rules()
        return jsonify(
            {rule.endpoint: rule.rule for rule in rules_iterator if rule.endpoint not in ('route_map', 'static')})

不过我们在这里是另辟蹊径，也为后续异常全局错误做铺垫，统一对 from werkzeug.exceptions import HTTPException 中的HTTPException进行重写一次。

PS:此处参考是7月老师的全局异常处理。

因为我们的是作为全局统一的，所以我们的可以把这个放在我们的项目Commom公告部分，方便其他模块的统一引入。

我们开始定义我们的全局异常错误基类：

其中我们的返回格式按之前确定的格式一样：

{
 code:int,   #返回状态码
 message:str, #返回信息描述
 result:object,  #返回结果内容信息
 success：bool, # 请求是否处理成功
 timestamp：str, # 请求响应时间戳
}

首先我们的需要定义基类对象common下的exc的__init__文件内进行编写我们的基类代码：

#!/usr/bin/evn python
# coding=utf-8

from flask import request
from werkzeug.exceptions import HTTPException
import time
import json
import datetime
import decimal


class CJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if hasattr(obj, 'keys') and hasattr(obj, '__getitem__'):
            return dict(obj)
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        if isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        if isinstance(obj, decimal.Decimal):
            return float(obj)
        if isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        return json.JSONEncoder.default(self, obj)


def dict_to_json(dict={}):
    return json.dumps(dict, cls=CJsonEncoder)


# 不格式化的输出ensure_ascii==false 输出中文的时候，保持中文的输出
def dict_to_json_ensure_ascii(dict={}, ensure_ascii=False):
    return json.dumps(dict, cls=CJsonEncoder, ensure_ascii=ensure_ascii)


# 格式化排版缩进输出-ensure_ascii==false 输出中文的时候，保持中文的输出
def dict_to_json_ensure_ascii_indent(dict={}, ensure_ascii=False):
    return json.dumps(dict, cls=CJsonEncoder, ensure_ascii=ensure_ascii, indent=4)


class ApiAdminJsonResponse(HTTPException):
    code = 500
    api_code = 999
    result = None  # 结果可以是{} 或 []
    message = '成功'
    success = True
    timestamp = int(time.time() * 1000)
    # 默认的支持的跨域处理
    customize_headers = {
        # 'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': '*,School-Teacher-Token,school-teacher-token,T-Access-Token,x-access-token,Referer, Accept, Origin, User-Agent,X-Requested-With, Content-Type, X-File-Name',
        'Access-Control-Request-Headers': 'Content-Type,Access-Token',
        'Content-Type': 'application/json;charset=UTF-8'
    }

    # 还可以自定义响应头信息
    def __init__(self, success=None, code=None, message=None, api_code=None, result=None, headers={}, is_limit_cors=False):

        if code:
            self.code = code
        if message:
            self.message = message
        if api_code or api_code == 0:
            self.api_code = api_code
        if result:
            self.result = result
        if headers:
            self.customize_headers.update(headers)
        if success != None:
            self.success = success
        # self.code = code
        # self.api_code = api_code
        # 是否禁止跨域请求处理
        self.is_limit_cors = is_limit_cors
        super(ApiAdminJsonResponse, self).__init__(message, None)

    def get_body(self, environ=None):

        body = dict(
            message=self.message,
            code=self.api_code,
            success=self.success,
            result=self.result,
            timestamp=self.timestamp,
            # 形如request="POST v1/client/register"
            # request=request.method + ' ' + self.get_url_no_param()
        )
        text = dict_to_json(body)
        return text

    # 重写响应的头部信息----可以自定义返回响应头部信息
    def get_headers(self, environ=None):
        if self.is_limit_cors:
            pass
            # del self.customize_headers['Access-Control-Allow-Origin']
            if 'Access-Control-Allow-Origin' in self.customize_headers.keys():
                del self.customize_headers['Access-Control-Allow-Origin']
        return self.customize_headers

    @staticmethod
    def get_url_no_param():
        full_url = str(request.full_path)
        main_path = full_url.split('?')
        return main_path[0]


其中CJsonEncoder的主要是为了处理，我们的部分的类型再转化JSON格式后无法兼容的异常问题而处理。

从上面的代码看我们的定义的对应字段描述如下：

字段 说明描述
code 对应HTTP的请求响应码如 200 或 500等响应码
api_code 自定义响应码 （就是格式中定义的code，因HTTPException也存在Code,为了区分使用这个APIcode来对应）
result 响应结果内容信息
message 请求结果的描述
success 请求最终结果是否成功 true 或 false
timestamp 请求响应的时间戳

上面基类中其实还扩展了对自定义响应报文头的内容，主要是用于自定义的时候返回的跨域问题处理,部分的浏览器似乎需要具体明细到你需要支持的哪些自定义的响应头才可以实现跨域，如UC浏览器。所以部分的跨域请求头，有我们的需要用到自定的请求头。

因为后续的我们的JTW的Token是放在请求头里的，需要我们的也定义一个x-access-token。

    # 默认的支持的跨域处理
    customize_headers = {
        # 'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': '*,School-Teacher-Token,school-teacher-token,T-Access-Token,x-access-token,Referer, Accept, Origin, User-Agent,X-Requested-With, Content-Type, X-File-Name',
        'Access-Control-Request-Headers': 'Content-Type,Access-Token',
        'Content-Type': 'application/json;charset=UTF-8'
    }

根据基类定义扩展其他错误异常类，扩展后的完整代码如下： 

#!/usr/bin/evn python
# coding=utf-8

from flask import request
from werkzeug.exceptions import HTTPException
import time
import json
import datetime
import decimal


class CJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if hasattr(obj, 'keys') and hasattr(obj, '__getitem__'):
            return dict(obj)
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        if isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        if isinstance(obj, decimal.Decimal):
            return float(obj)
        if isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        return json.JSONEncoder.default(self, obj)


def dict_to_json(dict={}):
    return json.dumps(dict, cls=CJsonEncoder)


# 不格式化的输出ensure_ascii==false 输出中文的时候，保持中文的输出
def dict_to_json_ensure_ascii(dict={}, ensure_ascii=False):
    return json.dumps(dict, cls=CJsonEncoder, ensure_ascii=ensure_ascii)


# 格式化排版缩进输出-ensure_ascii==false 输出中文的时候，保持中文的输出
def dict_to_json_ensure_ascii_indent(dict={}, ensure_ascii=False):
    return json.dumps(dict, cls=CJsonEncoder, ensure_ascii=ensure_ascii, indent=4)


def get_timestamp13():
    return int(time.time() * 1000)


class ApiAdminJsonResponse(HTTPException):
    code = 500
    api_code = 999
    result = None  # 结果可以是{} 或 []
    message = '成功'
    success = True
    timestamp = get_timestamp13()
    # 默认的支持的跨域处理
    customize_headers = {
        # 'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': '*,School-Teacher-Token,school-teacher-token,T-Access-Token,x-access-token,Referer, Accept, Origin, User-Agent,X-Requested-With, Content-Type, X-File-Name',
        'Access-Control-Request-Headers': 'Content-Type,Access-Token',
        'Content-Type': 'application/json;charset=UTF-8'
    }

    # 还可以自定义响应头信息
    def __init__(self, success=None, code=None, message=None, api_code=None, result=None, headers={}, is_limit_cors=False):

        if code:
            self.code = code
        if message:
            self.message = message
        if api_code or api_code == 0:
            self.api_code = api_code
        if result:
            self.result = result
        if headers:
            self.customize_headers.update(headers)
        if success != None:
            self.success = success
        # self.code = code
        # self.api_code = api_code
        # 是否禁止跨域请求处理
        self.is_limit_cors = is_limit_cors
        super(ApiAdminJsonResponse, self).__init__(message, None)

    def get_body(self, environ=None):

        body = dict(
            message=self.message,
            code=self.api_code,
            success=self.success,
            result=self.result,
            timestamp=self.timestamp,
            # 形如request="POST v1/client/register"
            # request=request.method + ' ' + self.get_url_no_param()
        )
        text = dict_to_json(body)
        return text

    # 重写响应的头部信息----可以自定义返回响应头部信息
    def get_headers(self, environ=None):
        if self.is_limit_cors:
            pass
            # del self.customize_headers['Access-Control-Allow-Origin']
            if 'Access-Control-Allow-Origin' in self.customize_headers.keys():
                del self.customize_headers['Access-Control-Allow-Origin']
        return self.customize_headers

    @staticmethod
    def get_url_no_param():
        full_url = str(request.full_path)
        main_path = full_url.split('?')
        return main_path[0]


class BadrequestException(ApiAdminJsonResponse):
    code = 400
    api_code = 10031
    result = None  # 结果可以是{} 或 []
    message = '错误的请求'
    success = False
    timestamp = get_timestamp13()


class LimiterResException(ApiAdminJsonResponse):
    code = 429
    api_code = 429
    result = None  # 结果可以是{} 或 []
    message = '访问的速度过快'
    success = False
    timestamp = get_timestamp13()


class ParameterException(ApiAdminJsonResponse):
    code = 400
    result = {}
    message = '参数校验错误,请检查提交的参数信息'
    api_code = 10031
    success = False
    timestamp = get_timestamp13()


class UnauthorizedException(ApiAdminJsonResponse):
    code = 401
    result = {}
    message = '未经许可授权'
    api_code = 10032
    success = False
    timestamp = get_timestamp13()


class ForbiddenException(ApiAdminJsonResponse):
    code = 403
    result = {}
    message = '失败！当前访问没有权限，或操作的数据没权限!'
    api_code = 10033
    success = False
    timestamp = get_timestamp13()


class NotfoundException(ApiAdminJsonResponse):
    code = 404
    result = {}
    message = '访问地址不存在'
    api_code = 10034
    success = False
    timestamp = get_timestamp13()


class MethodnotallowedException(ApiAdminJsonResponse):
    code = 405
    result = {}
    message = '不允许使用此方法提交访问'
    api_code = 10034
    success = False
    timestamp = get_timestamp13()


class OtherException(ApiAdminJsonResponse):
    code = 800
    result = {}
    message = '未知的其他HTTPEOOER异常'
    api_code = 10034
    success = False
    timestamp = get_timestamp13()


class InternalErrorException(ApiAdminJsonResponse):
    code = 500
    result = {}
    message = '程序员哥哥睡眠不足，系统崩溃了！'
    api_code = 500
    success = False
    timestamp = get_timestamp13()


class InvalidTokenException(ApiAdminJsonResponse):
    code = 401
    api_code = 401
    message = '很久没操作，令牌失效'
    success = False


class ExpiredTokenException(ApiAdminJsonResponse):
    code = 422
    message = '很久没操作，令牌过期'
    api_code = 10050
    success = False


class Success(ApiAdminJsonResponse):
    code = 200
    api_code = 200
    result = None  # 结果可以是{} 或 []
    message = '自定义成功返回'
    success = True


class Fail(ApiAdminJsonResponse):
    code = 200
    api_code = 200
    result = None  # 结果可以是{} 或 []
    message = '自定义成功返回'
    success = False


class FileTooLargeException(ApiAdminJsonResponse):
    code = 413
    api_code = 413
    result = None  # 结果可以是{} 或 []
    message = '文件体积过大'


class FileTooManyException(ApiAdminJsonResponse):
    code = 413
    message = '文件数量过多'
    api_code = 10120
    result = None  # 结果可以是{} 或 []

修改的app.py的启动，加一个测试路由测试我们的定义的返回信息：


if __name__ == '__main__':
    app = create_app()


    @app.route('/ceshi')
    def ceshi():
        from common.exc import Success
        return Success(api_code=0, result=None, message='添加成功!')


    app.run(host='127.0.0.1', port='8808', debug=True)

然后访问的地址http://127.0.0.1:8808/ceshi查看返回结果：

全局异常捕获处理

在app.py中我们的有一个是注册初始化了全局的错误异常函数：


def __init_error_handlers(app):
    pass

    # 自定义404错误码的返回信息异常的处理
    @app.errorhandler(Exception)
    def custom_error_handler(e):
        pass

其中Exception就是当我们的请求出现异常的时候会进入到这个函数里面进行处理。我们可以在这地方进行全局捕获处理。

注意我们需要引入包：from common.exc import *

然后完善我们的全局异常捕获处理函数：

def __init_error_handlers(app):
    pass

    # 自定义404错误码的返回信息异常的处理
    @app.errorhandler(Exception)
    def custom_error_handler(e):

        if isinstance(e, ApiAdminJsonResponse):
            # 直接的再次的返回，这样在From中也可以直接的返回了！
            # 如果还需要其他自定义的返回提示，可以再这里再进行判断，比如，我想直接返回200的响应体信息的话
            return e
        if isinstance(e, HTTPException):
            if e.code == 429:
                limt = LimiterResException()
                limt.msg = limt.msg + ' ' + e.description
                return limt
            if e.code == 404:
                return NotfoundException()
            if e.code == 404:
                return NotfoundException()

            # 其他错误处理------------------ 它跑到了全局了去
            code = e.code
            msg = e.description
            api_code = 20000
            return OtherException(message=msg, code=code, api_code=api_code)
        else:
            # 如果是Debug
            if not app.config['DEBUG']:
                return InternalErrorException()
            else:
                print("异常")
                # raise e
                return InternalErrorException()

测试抛出异常，定义个整除错误异常：

if __name__ == '__main__':
    app = create_app()


    @app.route('/ceshi')
    def ceshi():
        from common.exc import Success
        return Success(api_code=0, result=None, message='添加成功!')


    @app.route('/ceshi_exc')
    def ceshi_exc():

        4564/0

        from common.exc import Success
        return Success(api_code=0, result=None, message='添加成功!')

    app.run(host='127.0.0.1', port='8808', debug=True)

程序异常之后最终进入到了


            if not app.config['DEBUG']:
                return InternalErrorException()
            else:
                # 整除错误跑进了这里来了！！！
                print("异常")
                # raise e
                return InternalErrorException()

请求地址http://127.0.0.1:8808/ceshi_exc结果,正常进行统一返回了！！！

END

以上我们的完成统一API返回格式已经全局异常处理。

小钟同学 | 文  【原创】【转载请联系本人】| QQ：308711822


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： Flask + Ant Vue 实战（4）-脚手架配置文件读取解析
下一篇： Flask + Ant Vue 实战 （6）-定义中间件
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111698

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593