Python自定义异常类与API错误处理中间件实战

脚本专家

Python内置异常覆盖了大多数错误场景，但在电商、支付、API等应用中，ValueError往往说不清到底是库存不足、优惠券过期还是参数非法。自定义异常的本质就是继承Exception的普通类，用来表达特定于应用的错误。

一、创建自定义异常类

先看最小用法：为库存不足、支付失败、优惠券过期分别定义异常类。它们都继承Exception，命名以Error或Exception结尾。抛出时携带业务信息，捕获时就能精确处理。

2. class InsufficientStockError(Exception):
3.     '''库存不足异常'''
4.     pass
6. class PaymentFailedError(Exception):
7.     '''支付失败异常'''
8.     pass
10. class CouponExpiredError(Exception):
11.     '''优惠券过期异常'''
12.     pass
14. def place_order(product_id, quantity, stock):
15.     if quantity > stock:
16.         raise InsufficientStockError(
17.             f'库存不足: 需要{quantity}件，实际库存{stock}件'
18.         )
20. try:
21.     place_order('P001', 10, 5)
22. except InsufficientStockError as e:
23.     print(f'下单失败: {e}')

复制代码

这里的关键不是语法，而是语义：内置ValueError无法区分库存不足和价格非法，而InsufficientStockError可以。

二、设计异常层次结构

当项目规模变大时，不要散落定义异常。更合适的做法是建立层次结构：基础异常 -> 分类异常 -> 具体异常。基础异常MyAppError统一继承Exception；DataError、BusinessError、ExternalServiceError按错误来源分类；具体异常继承对应分类。

2. class MyAppError(Exception):
3.     '''应用的基础异常类'''
4.     pass
6. class DataError(MyAppError):
7.     '''数据相关错误'''
8.     pass
10. class BusinessError(MyAppError):
11.     '''业务逻辑错误'''
12.     pass
14. class ExternalServiceError(MyAppError):
15.     '''外部服务错误'''
16.     pass
18. class ValidationError(DataError):
19.     '''数据验证失败'''
20.     pass
22. class DuplicateUserError(DataError):
23.     '''用户已存在'''
24.     pass
26. class InsufficientBalanceError(BusinessError):
27.     '''余额不足'''
28.     pass
30. class PaymentTimeoutError(ExternalServiceError):
31.     '''支付超时'''
32.     pass

复制代码

这种结构有三个直接好处：可以按类别捕获，比如except DataError捕获所有数据错误；可以精确捕获，比如只处理DuplicateUserError；异常类名本身就是代码文档，阅读调用链时能看出可能出什么错。

三、让异常携带上下文

如果异常只带一条消息，API和日志很难定位问题。可以在异常类的__init__中接收field、value、code等参数，并提供一个to_dict方法，方便Web API返回结构化错误。

2. class ValidationError(Exception):
3.     '''数据验证错误——携带详细的字段信息'''
4.     def __init__(self, message, field=None, value=None, code=None):
5.         super().__init__(message)
6.         self.field = field
7.         self.value = value
8.         self.code = code
10.     def to_dict(self):
11.         '''转为字典——方便API返回'''
12.         return {
13.             'error': str(self),
14.             'field': self.field,
15.             'value': str(self.value) if self.value else None,
16.             'code': self.code or 'VALIDATION_ERROR',
17.         }
19. def validate_product(data):
20.     if not data.get('name'):
21.         raise ValidationError(
22.             '产品名称不能为空',
23.             field='name',
24.             code='MISSING_NAME'
25.         )
26.     if data.get('price', 0) <= 0:
27.         raise ValidationError(
28.             '产品价格必须大于0',
29.             field='price',
30.             value=data.get('price'),
31.             code='INVALID_PRICE'
32.         )
34. try:
35.     validate_product({'name': '产品A', 'price': -10})
36. except ValidationError as e:
37.     print(e.to_dict())

复制代码

输出会是：

2. {'error': '产品价格必须大于0', 'field': 'price', 'value': '-10', 'code': 'INVALID_PRICE'}

复制代码

注意，value在to_dict中转成了字符串，这样能安全放入JSON响应。code用于给前端或调用方提供稳定标识，不依赖错误消息文本。

四、支付系统中的异常体系

支付场景是自定义异常的典型场景。可以定义PaymentError作为基类，携带order_id和amount；PaymentDeclinedError增加reason；PaymentTimeoutError增加timeout_seconds；RefundError增加refund_id。支付服务调用网关时，根据响应状态抛出不同异常，超时则把TimeoutError转换成PaymentTimeoutError，并用from e保留原因链。

2. class PaymentError(Exception):
3.     '''支付相关错误的基类'''
4.     def __init__(self, message, order_id=None, amount=None):
5.         super().__init__(message)
6.         self.order_id = order_id
7.         self.amount = amount
9. class PaymentDeclinedError(PaymentError):
10.     '''支付被拒绝'''
11.     def __init__(self, message, order_id=None, amount=None, reason=None):
12.         super().__init__(message, order_id, amount)
13.         self.reason = reason
15. class PaymentTimeoutError(PaymentError):
16.     '''支付超时'''
17.     def __init__(self, message, order_id=None, amount=None, timeout_seconds=None):
18.         super().__init__(message, order_id, amount)
19.         self.timeout_seconds = timeout_seconds
21. class RefundError(PaymentError):
22.     '''退款失败'''
23.     def __init__(self, message, order_id=None, amount=None, refund_id=None):
24.         super().__init__(message, order_id, amount)
25.         self.refund_id = refund_id
27. class PaymentService:
28.     @staticmethod
29.     def pay(order_id, amount, payment_method):
30.         '''执行支付'''
31.         try:
32.             response = PaymentService._call_gateway(order_id, amount, payment_method)
33.             if response['status'] == 'declined':
34.                 reason_msg = response.get('message', '未知原因')
35.                 raise PaymentDeclinedError(
36.                     f'支付被拒绝: {reason_msg}',
37.                     order_id=order_id,
38.                     amount=amount,
39.                     reason=response.get('reason')
40.                 )
41.             return response
42.         except TimeoutError as e:
43.             raise PaymentTimeoutError(
44.                 '支付网关超时',
45.                 order_id=order_id,
46.                 amount=amount,
47.                 timeout_seconds=30
48.             ) from e
50.     @staticmethod
51.     def _call_gateway(order_id, amount, method):
52.         return {'status': 'declined', 'reason': 'insufficient_funds'}
54. service = PaymentService()
55. try:
56.     result = service.pay('ORD-001', 299.99, 'credit_card')
57. except PaymentDeclinedError as e:
58.     print(f'支付失败: {e}')
59.     print(f' 订单: {e.order_id}')
60.     print(f' 金额: {e.amount}')
61.     print(f' 原因: {e.reason}')
62. except PaymentTimeoutError as e:
63.     print(f'支付超时: {e}')
64. except PaymentError as e:
65.     print(f'其他支付错误: {e}')

复制代码

五、API异常处理中间件

Web API中常需要把异常统一转换成HTTP状态码和JSON结构。可以定义APIError基类，类属性status_code和error_code；子类BadRequestError、NotFoundError、UnauthorizedError、ForbiddenError分别设置400、404、401、403。再用装饰器handle_api_error包裹处理函数：遇到APIError时返回对应状态码和错误详情；遇到未预期异常时记录日志并返回500，避免把内部堆栈暴露给客户端。

2. class APIError(Exception):
3.     '''API错误的基类'''
4.     status_code = 500
5.     error_code = 'INTERNAL_ERROR'
6.     def __init__(self, message, details=None):
7.         super().__init__(message)
8.         self.details = details or {}
10. class BadRequestError(APIError):
11.     status_code = 400
12.     error_code = 'BAD_REQUEST'
14. class NotFoundError(APIError):
15.     status_code = 404
16.     error_code = 'NOT_FOUND'
18. class UnauthorizedError(APIError):
19.     status_code = 401
20.     error_code = 'UNAUTHORIZED'
22. class ForbiddenError(APIError):
23.     status_code = 403
24.     error_code = 'FORBIDDEN'
26. def handle_api_error(func):
27.     '''装饰器——统一处理API异常'''
28.     def wrapper(*args, **kwargs):
29.         try:
30.             return func(*args, **kwargs)
31.         except APIError as e:
32.             return {
33.                 'success': False,
34.                 'error': {
35.                     'code': e.error_code,
36.                     'message': str(e),
37.                     'details': e.details,
38.                 }
39.             }, e.status_code
40.         except Exception as e:
41.             import logging
42.             logging.exception('未预期的服务器错误')
43.             return {
44.                 'success': False,
45.                 'error': {
46.                     'code': 'INTERNAL_ERROR',
47.                     'message': '服务器内部错误',
48.                 }
49.             }, 500
50.     return wrapper
52. @handle_api_error
53. def get_user(user_id):
54.     '''获取用户——API处理函数'''
55.     if not isinstance(user_id, int) or user_id <= 0:
56.         raise BadRequestError(
57.             '无效的用户ID',
58.             details={'user_id': user_id}
59.         )
60.     user = find_user(user_id)
61.     if user is None:
62.         raise NotFoundError(
63.             f'用户{user_id}不存在',
64.             details={'user_id': user_id}
65.         )
66.     return {'success': True, 'data': user}
68. def find_user(user_id):
69.     return None
71. result, status = get_user(999)
72. print(f'状态码: {status}')
73. print(f'响应: {result}')

复制代码

调用get_user(999)时，find_user返回None，因此抛出NotFoundError，最终返回状态码404，响应中success为False，error.code为NOT_FOUND，details里带有user_id。这个模式适合在Flask、Django等框架之外先理解异常转换逻辑，再接入框架自己的错误处理器。

六、要点与适用场景

自定义异常的核心价值，是把“通用错误”升级为“领域专用错误”。实现时注意几点：继承Exception而不是BaseException，因为BaseException通常留给系统退出等信号；为项目建立基础异常->分类->具体异常的层次结构；异常类可以携带field、value、code、order_id、amount、timeout_seconds等上下文；命名以Error或Exception结尾。内置异常无法精确描述错误时，就应该创建自定义异常；好的异常命名本身就是文档。

适用场景包括电商下单、支付网关调用、API参数校验、用户查询、退款流程等。只要调用方需要根据错误类型走不同分支，或者需要向API返回稳定错误码，自定义异常体系就比单一ValueError更有维护价值。