Python 数据库异常处理实战：IntegrityError、OperationalError 与自定义异常

脚本专家

Python 操作数据库时，异常处理是绕不开的坑。无论是 SQLite、MySQL 还是 PostgreSQL，常见的错误可以分成连接层面、SQL 层面和数据完整性层面。只有针对不同异常采取不同策略，才能写出健壮的数据库代码。

一、连接层面的异常

数据库连接可能因为服务器宕机、超时、磁盘满等问题而失败。在 Python 的 sqlite3 中，这类异常统一为 sqlite3.OperationalError。

2. import sqlite3
3. try:
4.     conn = sqlite3.connect('production.db')
5. except sqlite3.OperationalError as e:
6.     print(f"数据库连接失败: {e}")
7.     # 可能的原因：文件打不开、权限不足、网络不通

复制代码

OperationalError 是与执行环境相关的错误，和 SQL 语句本身无关。常见的场景包括连接超时、文件被锁、磁盘满等。这类错误适合重试或走降级方案，而不是直接返回错误。

二、SQL 层面的异常（编程错误）

表不存在、列拼写错误、SQL 语法错误等都属于编程错误。在 SQLite 中这些异常也是 OperationalError（的一个子集），但在 MySQL 的 pymysql 中有专门的 ProgrammingError。

2. # 表不存在
3. cursor.execute("SELECT * FROM users")
4. # sqlite3.OperationalError: no such table: users
6. # 列不存在
7. cursor.execute("SELECT phone FROM user")
8. # sqlite3.OperationalError: no such column: phone
10. # SQL 语法错误
11. cursor.execute("CREATE TABLE users id INTEGER")
12. # sqlite3.OperationalError: near "id": syntax error

复制代码

ProgrammingError 意味着代码写错了，比如表名或列名拼写错误、SQL 语法有问题、参数绑定的数量不匹配。这类错误不能用重试解决，正确做法是修正代码。

三、数据完整性层面的异常

违反唯一约束、主键冲突、外键失败等会抛出 sqlite3.IntegrityError。典型场景是重复插入相同的手机号或邮箱。

2. cursor.execute("INSERT INTO users (phone) VALUES ('13800138000')")
3. # sqlite3.IntegrityError: UNIQUE constraint failed: users.phone

复制代码

IntegrityError 通常需要业务逻辑来处理，而不是简单打印日志后返回“服务器错误”。例如在用户注册场景中，收到 IntegrityError 时应当检查是否是手机号已存在，并返回“用户已存在”的业务提示。

四、业务场景里的正确姿势

以下代码演示了如何区分 IntegrityError 的类型，并分别处理：

2. def register_user(phone, name):
3.     try:
4.         conn.execute(
5.             "INSERT INTO users (phone, name) VALUES (?, ?)",
6.             (phone, name)
7.         )
8.         conn.commit()
9.         return {"status": "success", "msg": "注册成功"}
10.     except sqlite3.IntegrityError as e:
11.         if "UNIQUE constraint failed: users.phone" in str(e):
12.             return {"status": "exists", "msg": "该手机号已注册"}
13.         else:
14.             logger.error(f"IntegrityError: {e}")
15.             raise
16.     except sqlite3.OperationalError as e:
17.         logger.warning(f"数据库操作失败，准备重试: {e}")
18.         time.sleep(1)
19.         return register_user(phone, name)  # 递归重试（需设上限）

复制代码

核心原则：不要用一个笼统的 except Exception 吞掉所有异常，应该针对不同异常类型做精确处理。

五、上下文管理器：自动释放资源

手动管理连接和游标很容易忘记关闭，尤其在异常发生时导致资源泄漏。使用 with 语句可以自动处理：

2. def get_user(user_id):
3.     with sqlite3.connect('app.db') as conn:
4.         with conn.cursor() as cursor:
5.             cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
6.             return cursor.fetchone()
7.     # 退出 with 块时，cursor 和 conn 自动关闭，即使发生异常

复制代码

ORM 如 SQLAlchemy 也提供了类似的上下文管理机制。

六、自定义异常：让错误信息带上业务上下文

直接暴露数据库底层错误（如 "Table doesn't exist"）对用户没有意义。更好的做法是包装成业务异常：

2. class UserRegistrationError(Exception):
3.     pass
5. class UserAlreadyExistsError(UserRegistrationError):
6.     pass
8. class DatabaseConnectionError(UserRegistrationError):
9.     pass
11. def register_user(phone, name):
12.     try:
13.         # 数据库操作...
14.     except sqlite3.IntegrityError as e:
15.         if "UNIQUE constraint" in str(e):
16.             raise UserAlreadyExistsError(f"手机号{phone}已注册") from e
17.     except sqlite3.OperationalError as e:
18.         raise DatabaseConnectionError("数据库暂时不可用") from e

复制代码

上层代码可以直接按异常类型捕获：

2. try:
3.     result = register_user(phone, name)
4. except UserAlreadyExistsError:
5.     # 前端显示"用户已存在"
6. except DatabaseConnectionError:
7.     # 前端显示"系统繁忙，请稍后重试"

复制代码

使用 raise ... from e 可以保留原始异常栈，方便调试。

七、完整实战模板

下面是一个可复用的装饰器，用于自动重试临时性数据库错误，并结合自定义异常构建完整工具：

2. import sqlite3
3. import time
4. import logging
5. logger = logging.getLogger(__name__)
7. class DatabaseError(Exception):
8.     pass
10. class DuplicateRecordError(DatabaseError):
11.     pass
13. def with_retry(max_retries=3, delay=1):
14.     def decorator(func):
15.         def wrapper(*args, **kwargs):
16.             for attempt in range(max_retries):
17.                 try:
18.                     return func(*args, **kwargs)
19.                 except sqlite3.OperationalError as e:
20.                     if attempt == max_retries - 1:
21.                         raise DatabaseError(f"操作失败，已重试{max_retries}次") from e
22.                     logger.warning(f"第{attempt+1}次重试，错误: {e}")
23.                     time.sleep(delay * (attempt + 1))  # 指数退避
24.         return wrapper
25.     return decorator
27. @with_retry(max_retries=3)
28. def create_user(phone, name):
29.     try:
30.         with sqlite3.connect('app.db') as conn:
31.             conn.execute(
32.                 "INSERT INTO users (phone, name) VALUES (?, ?)",
33.                 (phone, name)
34.             )
35.             conn.commit()
36.             return {"id": conn.lastrowid, "phone": phone}
37.     except sqlite3.IntegrityError as e:
38.         if "UNIQUE constraint" in str(e):
39.             raise DuplicateRecordError(f"手机号{phone}已存在") from e
40.         raise DatabaseError(f"数据完整性错误: {e}") from e

复制代码

八、总结三条原则

1. 区分异常类型：OperationalError 重试，IntegrityError 走业务逻辑，ProgrammingError 改代码。
2. 用上下文管理器：数据库连接和游标用 with 管理，避免资源泄漏。
3. 包装成业务异常：让上层代码按业务逻辑处理，而不是处理数据库底层错误。

按照这套模板重构数据库操作后，网络抖动、重复提交等问题都能被正确处理——要么重试成功，要么返回明确的业务提示，避免半夜报警。

热心网友5

感谢楼主分享这么详细的数据库异常处理实战经验！尤其是区分 OperationalError、ProgrammingError 和 IntegrityError 的典型场景，对新手来说非常清晰。我特别赞同“不要用一个笼统的 except Exception 吞掉所有异常”这个原则，很多代码就是在这上面埋坑的。 有一个小问题想请教：在注册用户递归重试的例子中，如果遇到 OperationalError（比如连接超时），递归重试没有设置上限，会不会导致无限递归？通常我会加一个重试次数参数，或者用循环代替递归，不知道楼主怎么看？另外，自定义异常这部分我觉得很实用，但有时候在大型项目里，不同模块的异常类会很多，有没有推荐的命名或组织方式？

热心网友5

楼主的总结非常到位，尤其把 OperationalError 和 IntegrityError 区分开处理这点，在实际项目中太重要了。以前踩过坑，一个 try except 全吞掉，结果 IntegrityError 被当成普通错误返回“系统繁忙”，用户反复重试也注册不了，后来才改成按异常类型分支处理。 想请教一下：对于 PostgreSQL 的 psycopg2，它的 IntegrityError 下还有更细的 UniqueViolation 和 ForeignKeyViolation 子类，这样是不是可以直接捕捉子类而不需要做字符串匹配？另外自定义异常类里，是否需要保留原异常栈信息（比如用 `raise ... from e`）方便调试？期待楼主分享更多实战经验。

热心网友5

感谢楼主的详细分享，把数据库异常处理拆成连接、SQL、数据完整性三个层面讲得很清楚，尤其是用具体的 sqlite3 场景做对比，容易理解。实际项目中确实很容易一把 except 吞掉所有错误，导致问题很难排查。你提到的根据异常特征字符串区分 IntegrityError 的写法很实用，这样就能对用户给出明确提示，而不是笼统的“系统错误”。自定义异常类加上业务上下文，也让调用方处理起来更优雅。最后 with 语句那段也是基础但容易忽略的细节。收藏了，以后写数据库操作时当模板参考。