python中通過Django捕獲所有異常的處理

更新時間：2021年09月29日 17:16:09 作者：憶想不到的暉

誠然，每個人都會寫bug，程序拋異常是一件很正常的事；既然異?？偸菚?，那就想辦法在拋出后，盡早解決才是王道。不能老是等待用戶反饋異常和問題，萬一用戶懶得反饋了，豈不很尷尬

概述

在項目中統(tǒng)一異常處理，可以防止代碼中有未捕獲的異常出現(xiàn)。本文介紹如何在 Django 項目中進行統(tǒng)一異常的處理，再結(jié)合狀態(tài)碼枚舉類對項目異常信息進行日志記錄。

Django 統(tǒng)一異常處理

在 Django 項目中可以自定義中間件類繼承 django.middleware.common 下的 MiddlewareMixin 中間件類，重寫 process_exception 方法的異常處理邏輯，然后在項目配置下的中間件中注冊即可進行全局異常處理。

我是在項目自定義的 utils 包下 middlewares.py 模塊中下進行中間件的編寫。

# middlewares.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 項目中間件模塊 }
# @Date: 2021/09/24 8:18
from django.middleware.common import MiddlewareMixin



class ExceptionMiddleware(MiddlewareMixin):
    """統(tǒng)一異常處理中間件"""

    def process_exception(self, request, exception):
        """
        統(tǒng)一異常處理
        :param request: 請求對象
        :param exception: 異常對象
        :return:
        """
        # 異常處理
        print(exception)
        return None

這里暫時先簡單進行異常輸出，來模擬異常處理。最后不要忘記在配置文件中注冊中間件。django 項目默認的配置文件是 settings.py 我這里只是把配置文件單獨放到了 settings 包下然后改了文件名。

統(tǒng)一異常處理中間件配置

process_exception 方法介紹

process_exception 方法只有在視圖函數(shù)中出現(xiàn)異常了才執(zhí)行。該方法的返回值可以是一個 None 也可以是一個 HttpResponse 對象。

返回值是 None，頁面會報 500 狀態(tài)碼錯誤，視圖函數(shù)不會執(zhí)行。
返回值是 HttpResponse 對象，則是對應(yīng)的響應(yīng)信息，頁面不會報錯。

中間件中的方法

方法	作用
`process_request(self,request)`	在視圖函數(shù)之前執(zhí)行
`process_view(self, request, view_func, view_args, view_kwargs)`	視圖函數(shù)之前，`process_request` 方法之后執(zhí)行
`process_exception(self, request, exception)`	視圖函數(shù)中出現(xiàn)異常了才執(zhí)行
`process_response(self, request, response)`	視圖函數(shù)之后執(zhí)行

下面一圖就能比較好的呈現(xiàn) django 整個處理流程邏輯

更多的中間件細節(jié)可以去 Django 官方文檔進行了解。

統(tǒng)一異常處理具體設(shè)計

結(jié)合自定義的異常和狀態(tài)碼枚舉類，進行異常日志信息和業(yè)務(wù)邏輯的處理。

自定義異常模塊

# exceptions.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 項目異常模塊 }
# @Date: 2021/09/24 8:14

class CommonException(Exception):
    """公共異常類"""

    def __init__(self, enum_cls):
        self.code = enum_cls.code
        self.errmsg = enum_cls.errmsg
        self.enum_cls = enum_cls	# 狀態(tài)碼枚舉類
        super().__init__()


class BusinessException(CommonException):
    """業(yè)務(wù)異常類"""
    pass


class APIException(CommonException):
    """接口異常類"""
    pass

自定義狀態(tài)碼枚舉類

# enums.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 項目枚舉類模塊 }
# @Date: 2021/09/23 23:37

from enum import Enum


class StatusCodeEnum(Enum):
    """狀態(tài)碼枚舉類"""

    OK = (0, '成功')
    ERROR = (-1, '錯誤')
    SERVER_ERR = (500, '服務(wù)器異常')

    IMAGE_CODE_ERR = (4001, '圖形驗證碼錯誤')
    THROTTLING_ERR = (4002, '訪問過于頻繁')
    NECESSARY_PARAM_ERR = (4003, '缺少必傳參數(shù)')
    USER_ERR = (4004, '用戶名錯誤')
    PWD_ERR = (4005, '密碼錯誤')
    CPWD_ERR = (4006, '密碼不一致')
    MOBILE_ERR = (4007, '手機號錯誤')
    SMS_CODE_ERR = (4008, '短信驗證碼有誤')
    ALLOW_ERR = (4009, '未勾選協(xié)議')
    SESSION_ERR = (4010, '用戶未登錄')
    REGISTER_FAILED_ERR = (4011, '注冊失敗')

    DB_ERR = (5000, '數(shù)據(jù)庫錯誤')
    EMAIL_ERR = (5001, '郵箱錯誤')
    TEL_ERR = (5002, '固定電話錯誤')
    NODATA_ERR = (5003, '無數(shù)據(jù)')
    NEW_PWD_ERR = (5004, '新密碼錯誤')
    OPENID_ERR = (5005, '無效的openid')
    PARAM_ERR = (5006, '參數(shù)錯誤')
    STOCK_ERR = (5007, '庫存不足')

    @property
    def code(self):
        """獲取狀態(tài)碼"""
        return self.value[0]

    @property
    def errmsg(self):
        """獲取狀態(tài)碼信息"""
        return self.value[1]

自定義的異常類用于區(qū)分系統(tǒng)異常和業(yè)務(wù)來進行單獨處理。
狀態(tài)碼枚舉則是用來記錄對應(yīng)的異常信息。

狀態(tài)碼枚舉類的設(shè)計可以查閱巧用Python 枚舉類設(shè)計狀態(tài)碼信息

響應(yīng)信息統(tǒng)一結(jié)果的封裝

統(tǒng)一前后端交互數(shù)據(jù)和異常信息結(jié)果。

# result.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 項目信息返回結(jié)果模塊 }
# @Date: 2021/09/23 22:10
from .enums import StatusCodeEnum


class R(object):
    """
    統(tǒng)一項目信息返回結(jié)果類
    """

    def __init__(self):
        self.code = None
        self.errmsg = None
        self._data = dict()

    @staticmethod
    def ok():
        """
        組織成功響應(yīng)信息
        :return:
        """
        r = R()
        r.code = StatusCodeEnum.OK.code
        r.errmsg = StatusCodeEnum.OK.errmsg
        return r

    @staticmethod
    def error():
        """
        組織錯誤響應(yīng)信息
        :return:
        """
        r = R()
        r.code = StatusCodeEnum.ERROR.code
        r.errmsg = StatusCodeEnum.ERROR.errmsg
        return r

    @staticmethod
    def server_error():
        """
        組織服務(wù)器錯誤信息
        :return:
        """
        r = R()
        r.code = StatusCodeEnum.SERVER_ERR.code
        r.errmsg = StatusCodeEnum.SERVER_ERR.errmsg
        return r

    @staticmethod
    def set_result(enum):
        """
        組織對應(yīng)枚舉類的響應(yīng)信息
        :param enum: 狀態(tài)枚舉類
        :return:
        """
        r = R()
        r.code = enum.code
        r.errmsg = enum.errmsg
        return r

    def data(self, key=None, obj=None):
        """統(tǒng)一后端返回的數(shù)據(jù)"""

        if key:
            self._data[key] = obj

        context = {
            'code': self.code,
            'errmsg': self.errmsg,
            'data': self._data
        }
        return context

完善統(tǒng)一異常處理邏輯

# middlewares.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @Desc: { 項目中間件模塊 }
# @Date: 2021/09/24 8:18
import logging

from django.db import DatabaseError
from django.http.response import JsonResponse
from django.http import HttpResponseServerError
from django.middleware.common import MiddlewareMixin

from meiduo_mall.utils.result import R
from meiduo_mall.utils.enums import StatusCodeEnum
from meiduo_mall.utils.exceptions import BusinessException

logger = logging.getLogger('django')


class ExceptionMiddleware(MiddlewareMixin):
    """統(tǒng)一異常處理中間件"""

    def process_exception(self, request, exception):
        """
        統(tǒng)一異常處理
        :param request: 請求對象
        :param exception: 異常對象
        :return:
        """
        if isinstance(exception, BusinessException):
            # 業(yè)務(wù)異常處理
            data = R.set_result(exception.enum_cls).data()
            return JsonResponse(data)

        elif isinstance(exception, DatabaseError):
            # 數(shù)據(jù)庫異常
            r = R.set_result(StatusCodeEnum.DB_ERR)
            logger.error(r.data(), exc_info=True)
            return HttpResponseServerError(StatusCodeEnum.SERVER_ERR.errmsg)

        elif isinstance(exception, Exception):
            # 服務(wù)器異常處理
            r = R.server_error()
            logger.error(r.data(), exc_info=True)
            return HttpResponseServerError(r.errmsg)
        
        return None

應(yīng)用場景

注冊校驗

讓我們來看一段注冊校驗功能業(yè)務(wù)邏輯

	def verify_params(self, request):
        """
        校驗注冊信息
        :param request: 注冊請求對象
        :return: response_ret
        """
        # 接受參數(shù)
        self.username = request.POST.get('username')
        self.password = request.POST.get('password')
        self.confirm_pwd = request.POST.get('confirm_pwd')
        self.mobile = request.POST.get('mobile')
        self.allow = request.POST.get('allow')   

        if not all(all_args):
            # raise BusinessException(StatusCodeEnum.PARAM_ERR)
            response_ret = http.HttpResponseForbidden('參數(shù)錯誤')
            return response_ret

        # 用戶名 5-20個字符
        if not re.match(r'^[a-zA-Z0-9_]{5,20}', self.username):
            response_ret = http.HttpResponseForbidden('用戶名不規(guī)范')
            return response_ret

        # 密碼 8-20個字符
        if not re.match(r'^[a-zA-Z0-9]{8,20}', self.password):
            response_ret = http.HttpResponseForbidden('密碼不規(guī)范')
            return response_ret

        # 兩次密碼一致性
        if self.password != self.confirm_pwd:
            response_ret = http.HttpResponseForbidden('兩次密碼不一致')
            return response_ret

        # 手機號合法性
        if not re.match(r'^1[3-9]\d{9}$', self.mobile):
            response_ret = http.HttpResponseForbidden('手機號碼不合法')
            return response_ret

        # 是否勾選用戶協(xié)議
        if self.allow != 'on':
            response_ret = http.HttpResponseForbidden('請勾選用戶協(xié)議')
            return response_ret

        return response_ret

通過拋異常和設(shè)置狀態(tài)碼枚舉來處理

    def verify_params(self, request):
        """
        校驗注冊信息
        :param request: 注冊請求對象
        :return: response_ret
        """
        # 接受參數(shù)
        self.username = request.POST.get('username')
        self.password = request.POST.get('password')
        self.confirm_pwd = request.POST.get('confirm_pwd')
        self.mobile = request.POST.get('mobile')
        self.allow = request.POST.get('allow')

        # 校驗參數(shù)
        all_args = [self.username, self.password, self.confirm_pwd, self.mobile, self.allow]
        if not all(all_args):
            raise BusinessException(StatusCodeEnum.PARAM_ERR)

        # 用戶名 5-20個字符
        if not re.match(r'^[a-zA-Z0-9_]{5,20}', self.username):
            raise BusinessException(StatusCodeEnum.USER_ERR)

        # 密碼 8-20個字符
        if not re.match(r'^[a-zA-Z0-9]{8,20}', self.password):
            raise BusinessException(StatusCodeEnum.PWD_ERR)

        # 兩次密碼一致性
        if self.password != self.confirm_pwd:
            raise BusinessException(StatusCodeEnum.CPWD_ERR)

        # 手機號合法性
        if not re.match(r'^1[3-9]\d{9}$', self.mobile):
            raise BusinessException(StatusCodeEnum.MOBILE_ERR)

        # 是否勾選用戶協(xié)議
        if self.allow != 'on':
            raise BusinessException(StatusCodeEnum.ALLOW_ERR)

減少 try ... except ... 代碼塊

例如在對數(shù)據(jù)庫進行操作時，為了防止數(shù)據(jù)庫發(fā)生了意外的異常導(dǎo)致系統(tǒng)崩潰，通常加上 try ... except ...來記錄異常信息。然而配置了全局異常處理，則可以不用管理。

# 創(chuàng)建用戶
try:
    user = User.objects.create_user(
        username=self.username,
        password=self.password,
        mobile=self.mobile,
    )
except DatabaseError as e:
    logger.error(e)
    
    
# 有了全局的異常處理
user = User.objects.create_user(
        username=self.username,
        password=self.password,
        mobile=self.mobile,
    )

注意：如果需要通過異常捕獲來處理一些業(yè)務(wù)信息，則不可避免，如事務(wù)回滾等