异常处理（二）：raise主动抛异常，自定义异常类

——一个老架构师的“契约式编程”实战：用异常构建清晰、可维护的 KES 数据服务边界

---

开场白：别再用 return None 来“假装没事”了！

看看这段代码，是不是很眼熟？

def get_user_from_kes(user_id):
    if not user_id:
        return None  # 参数错了？用户不存在？还是数据库挂了？
    
    try:
        cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cur.fetchone()
    except Exception:
        return None  # 又一个 None！

调用方拿到 None，一脸懵逼：

• 是用户 ID 无效？
• 是用户根本不存在？
• 还是电科金仓数据库连不上？

在核心系统里，模糊的返回值比异常更危险——它让 bug 潜伏，让故障蔓延。

今天，咱们就聊聊 raise 主动抛异常 和 自定义异常类，教你用异常建立 清晰的服务契约，让每一行 KES 交互都“责任分明”。

---

一、为什么需要主动抛异常？因为“沉默是金”在这里是毒药

场景：参数校验

# ❌ 错误示范：静默失败
def create_order(user_id, amount):
    if not user_id or amount <= 0:
        return False  # 调用方：？？？

# ✅ 正确姿势：主动抛异常
def create_order(user_id, amount):
    if not user_id:
        raise ValueError("user_id 不能为空")
    if amount <= 0:
        raise ValueError("订单金额必须大于 0")
    
    # 继续执行...

💡 关键思想：
函数入口就是“关卡”，不合格的输入，当场拦截，绝不放行。

---

二、自定义异常类：给你的 KES 服务穿上“制服”

Python 内置异常（如 ValueError）太泛，无法表达业务语义。
自定义异常，就是给错误“贴标签”。

基础模板：

class KESBaseError(Exception):
    """电科金仓服务基类异常"""
    pass

class UserNotFoundError(KESBaseError):
    """用户不存在"""
    def __init__(self, user_id):
        self.user_id = user_id
        super().__init__(f"用户 {user_id} 在 KES 中不存在")

class InsufficientBalanceError(KESBaseError):
    """余额不足"""
    def __init__(self, user_id, balance, required):
        self.user_id = user_id
        self.balance = balance
        self.required = required
        super().__init__(
            f"用户 {user_id} 余额 {balance} 不足，需 {required}"
        )

class KESConnectionError(KESBaseError):
    """KES 连接异常（封装底层驱动异常）"""
    pass

优势：

• 调用方可以精确捕获：

  try:
      transfer_money(from_id, to_id, 1000)
  except UserNotFoundError as e:
      return {"error": "USER_NOT_FOUND", "user_id": e.user_id}
  except InsufficientBalanceError as e:
      return {
          "error": "INSUFFICIENT_BALANCE",
          "balance": e.balance,
          "required": e.required
      }
  

• 日志自带上下文：

  logger.error(f"转账失败: {e}")  # 自动打印完整错误信息
  

📌 驱动提示：确保使用电科金仓官方驱动以获得标准异常类型
👉 https://www.kingbase.com.cn/download.html#drive

---

三、实战：构建一个“异常契约清晰”的 KES 用户服务

import ksycopg2
from typing import Optional

# 自定义异常（放在 errors.py）
class KESUserError(Exception): pass
class UserNotFound(KESUserError): 
    def __init__(self, user_id): 
        super().__init__(f"用户 {user_id} 不存在")
        self.user_id = user_id

class DuplicateUser(KESUserError): 
    def __init__(self, user_id): 
        super().__init__(f"用户 {user_id} 已存在")
        self.user_id = user_id

# 核心服务
class KESUserService:
    def __init__(self, dsn: str):
        self.dsn = dsn
    
    def _get_connection(self):
        try:
            return ksycopg2.connect(self.dsn)
        except ksycopg2.OperationalError as e:
            raise KESConnectionError(f"无法连接电科金仓: {e}")
    
    def get_user(self, user_id: int) -> dict:
        """获取用户，不存在则抛异常"""
        if not isinstance(user_id, int) or user_id <= 0:
            raise ValueError("user_id 必须是正整数")
        
        with self._get_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id, name, balance FROM users WHERE id = %s", 
                    (user_id,)
                )
                row = cur.fetchone()
                if row is None:
                    raise UserNotFound(user_id)  # 主动抛出！
                
                return {"id": row[0], "name": row[1], "balance": row[2]}
    
    def create_user(self, user_id: int, name: str) -> None:
        """创建用户，已存在则抛异常"""
        if not name.strip():
            raise ValueError("用户名不能为空")
        
        try:
            with self._get_connection() as conn:
                with conn.cursor() as cur:
                    cur.execute(
                        "INSERT INTO users (id, name) VALUES (%s, %s)",
                        (user_id, name)
                    )
                    conn.commit()
        except ksycopg2.IntegrityError as e:
            if "unique_violation" in str(e):
                raise DuplicateUser(user_id)
            raise  # 其他 IntegrityError 重新抛出

调用方代码（清晰明了）：

service = KESUserService(dsn)

try:
    user = service.get_user(123)
    print(f"找到用户: {user['name']}")
except UserNotFound:
    print("用户不存在，跳过处理")
except KESConnectionError:
    print("KES 服务暂时不可用，请稍后重试")

✅ 效果：

• 错误类型明确
• 处理逻辑分离
• 无需猜测 None 的含义

---

四、高级技巧：异常链（Exception Chaining）

有时候，你想保留原始异常信息，同时抛出业务异常：

def execute_kes_query(sql):
    try:
        with ksycopg2.connect(dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(sql)
                return cur.fetchall()
    except ksycopg2.DatabaseError as e:
        # 保留原始异常（__cause__）
        raise KESQueryError(f"KES 查询失败: {sql}") from e

# 日志会显示：
# KESQueryError: KES 查询失败: SELECT * FROM xxx
# Caused by: psycopg2.ProgrammingError: table "xxx" does not exist

💡 用 raise ... from ... 实现异常链，既提供业务上下文，又保留技术细节。

---

五、避坑指南：自定义异常的 3 个雷区

❌ 雷区1：继承 Exception 而不是更具体的基类

# 不好
class MyError(Exception): pass

# 更好：按模块/功能分组
class KESDataError(Exception): pass
class KESAuthError(Exception): pass

❌ 雷区2：异常类不带上下文数据

# 不好
raise UserNotFound()

# 好：带上关键数据
raise UserNotFound(user_id=123)

❌ 雷区3：在 except 里吞掉原始异常

try:
    ...
except Exception:
    raise MyError("出错了")  # 原始异常信息丢失！

# 正确：用 raise ... from ...
except Exception as e:
    raise MyError("出错了") from e

---

六、特别提醒：电科金仓与异常设计哲学

1. KES 是企业级数据库，你的代码也该有企业级健壮性
   电科金仓支持 金融级高可用（RPO=0, RTO≈0），但如果你的代码用 return None 掩盖问题，再强的数据库也救不了业务逻辑。
   了解 KES 能力：https://kingbase.com.cn/product/details_549_476.html

2. 异常是 API 的一部分
   就像 KES 的 SQL 语法有规范一样，你的函数抛什么异常，也应该写在文档里（或通过类型注解）。

3. 监控要覆盖自定义异常
   在 APM 系统（如 SkyWalking）中，为 UserNotFound、InsufficientBalanceError 设置独立告警，快速定位业务问题。

---

结语：异常不是错误，是沟通的语言

在电科金仓这样的核心系统里，异常处理不是补丁，是设计。

好的异常设计：

• 让调用方知道“发生了什么”
• 让运维知道“哪里出了问题”
• 让开发者知道“该怎么修复”

下次写函数前，问自己：

“如果这里出错，我该抛什么异常，才能让调用方一眼看懂？”

如果答案是“抛个 ValueError”——
赶紧建个自定义异常类，给你的服务穿上专业的“制服”。

---

作者：一个坚信“异常是契约，不是噪音”的技术架构师
环境：Python 3.10 + ksycopg2 + 电科金仓 KES V9R1（支撑银行核心交易系统）
注：所有代码均来自生产实践，拒绝“玩具示例”！✅