Python异常处理之自定义异常类的创建与使用

# 自定义异常——就是继承exception的普通类
class insufficientstockerror(exception):
    """库存不足异常"""
    pass

class paymentfailederror(exception):
    """支付失败异常"""
    pass

class couponexpirederror(exception):
    """优惠券过期异常"""
    pass

# 使用
def place_order(product_id, quantity, stock):
    if quantity > stock:
        raise insufficientstockerror(
            f"库存不足: 需要{quantity}件，实际库存{stock}件"
        )

try:
    place_order("p001", 10, 5)
except insufficientstockerror as e:
    print(f"下单失败: {e}")

# ✅ 为你的项目设计异常层次结构

# 基础异常——项目的所有异常都继承它
class myapperror(exception):
    """应用的基础异常类"""

# 分类异常
class dataerror(myapperror):
    """数据相关错误"""

class businesserror(myapperror):
    """业务逻辑错误"""

class externalserviceerror(myapperror):
    """外部服务错误"""

# 具体异常
class validationerror(dataerror):
    """数据验证失败"""

class duplicateusererror(dataerror):
    """用户已存在"""

class insufficientbalanceerror(businesserror):
    """余额不足"""

class paymenttimeouterror(externalserviceerror):
    """支付超时"""

# 💡 异常层次结构的优势：
# 1. 可以按类别捕获：except dataerror —— 捕获所有数据错误
# 2. 可以精确捕获：except duplicateusererror —— 只捕获重复用户
# 3. 代码自文档——一眼看出可能出什么错

class validationerror(exception):
    """数据验证错误——携带详细的字段信息"""
    def __init__(self, message, field=none, value=none, code=none):
        super().__init__(message)
        self.field = field      # 哪个字段出错了
        self.value = value      # 导致错误的值
        self.code = code        # 错误码

    def to_dict(self):
        """转为字典——方便api返回"""
        return {
            "error": str(self),
            "field": self.field,
            "value": str(self.value) if self.value else none,
            "code": self.code or "validation_error",
        }

# 使用
def validate_product(data):
    if not data.get("name"):
        raise validationerror(
            "产品名称不能为空",
            field="name",
            code="missing_name"
        )
    if data.get("price", 0) <= 0:
        raise validationerror(
            "产品价格必须大于0",
            field="price",
            value=data.get("price"),
            code="invalid_price"
        )

try:
    validate_product({"name": "产品a", "price": -10})
except validationerror as e:
    print(e.to_dict())
    # {'error': '产品价格必须大于0', 'field': 'price', 'value': '-10', 'code': 'invalid_price'}

class paymenterror(exception):
    """支付相关错误的基类"""
    def __init__(self, message, order_id=none, amount=none):
        super().__init__(message)
        self.order_id = order_id
        self.amount = amount

class paymentdeclinederror(paymenterror):
    """支付被拒绝"""
    def __init__(self, message, order_id=none, amount=none, reason=none):
        super().__init__(message, order_id, amount)
        self.reason = reason

class paymenttimeouterror(paymenterror):
    """支付超时"""
    def __init__(self, message, order_id=none, amount=none, timeout_seconds=none):
        super().__init__(message, order_id, amount)
        self.timeout_seconds = timeout_seconds

class refunderror(paymenterror):
    """退款失败"""
    def __init__(self, message, order_id=none, amount=none, refund_id=none):
        super().__init__(message, order_id, amount)
        self.refund_id = refund_id

# 支付服务
class paymentservice:
    @staticmethod
    def pay(order_id, amount, payment_method):
        """执行支付"""
        try:
            # 调用外部支付网关
            response = paymentservice._call_gateway(order_id, amount, payment_method)
            if response["status"] == "declined":
                raise paymentdeclinederror(
                    f"支付被拒绝: {response.get('message', '未知原因')}",
                    order_id=order_id,
                    amount=amount,
                    reason=response.get("reason")
                )
            return response
        except timeouterror as e:
            raise paymenttimeouterror(
                "支付网关超时",
                order_id=order_id,
                amount=amount,
                timeout_seconds=30
            ) from e

    @staticmethod
    def _call_gateway(order_id, amount, method):
        # 模拟支付网关调用
        return {"status": "declined", "reason": "insufficient_funds"}

# 使用
service = paymentservice()
try:
    result = service.pay("ord-001", 299.99, "credit_card")
except paymentdeclinederror as e:
    print(f"支付失败: {e}")
    print(f"  订单: {e.order_id}")
    print(f"  金额: {e.amount}")
    print(f"  原因: {e.reason}")
except paymenttimeouterror as e:
    print(f"支付超时: {e}")
except paymenterror as e:
    print(f"其他支付错误: {e}")

# 模拟web框架中的异常处理

class apierror(exception):
    """api错误的基类"""
    status_code = 500
    error_code = "internal_error"

    def __init__(self, message, details=none):
        super().__init__(message)
        self.details = details or {}

class badrequesterror(apierror):
    status_code = 400
    error_code = "bad_request"

class notfounderror(apierror):
    status_code = 404
    error_code = "not_found"

class unauthorizederror(apierror):
    status_code = 401
    error_code = "unauthorized"

class forbiddenerror(apierror):
    status_code = 403
    error_code = "forbidden"

# 全局异常处理中间件
def handle_api_error(func):
    """装饰器——统一处理api异常"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except apierror as e:
            # 业务错误——返回对应的状态码
            return {
                "success": false,
                "error": {
                    "code": e.error_code,
                    "message": str(e),
                    "details": e.details,
                }
            }, e.status_code
        except exception as e:
            # 未预期的错误——500
            import logging
            logging.exception("未预期的服务器错误")
            return {
                "success": false,
                "error": {
                    "code": "internal_error",
                    "message": "服务器内部错误",
                }
            }, 500
    return wrapper

@handle_api_error
def get_user(user_id):
    """获取用户——api处理函数"""
    if not isinstance(user_id, int) or user_id <= 0:
        raise badrequesterror(
            "无效的用户id",
            details={"user_id": user_id}
        )

    user = find_user(user_id)
    if user is none:
        raise notfounderror(
            f"用户{user_id}不存在",
            details={"user_id": user_id}
        )

    return {"success": true, "data": user}

def find_user(user_id):
    return none  # 模拟用户不存在

# 测试
result, status = get_user(999)
print(f"状态码: {status}")
print(f"响应: {result}")
# 状态码: 404
# 响应: {'success': false, 'error': {'code': 'not_found', ...}}