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

脚本专家 发表于 2026-7-1 09:00:04

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

一、连接层面的异常

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

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

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

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

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

# 表不存在
cursor.execute("SELECT * FROM users")
# sqlite3.OperationalError: no such table: users

# 列不存在
cursor.execute("SELECT phone FROM user")
# sqlite3.OperationalError: no such column: phone

# SQL 语法错误
cursor.execute("CREATE TABLE users id INTEGER")
# sqlite3.OperationalError: near "id": syntax error

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

class UserRegistrationError(Exception):
pass

class UserAlreadyExistsError(UserRegistrationError):
pass

class DatabaseConnectionError(UserRegistrationError):
pass

def register_user(phone, name):
try:
   # 数据库操作...
except sqlite3.IntegrityError as e:
   if "UNIQUE constraint" in str(e):
         raise UserAlreadyExistsError(f"手机号{phone}已注册") from e
except sqlite3.OperationalError as e:
   raise DatabaseConnectionError("数据库暂时不可用") from e

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

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

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

七、完整实战模板

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

import sqlite3
import time
import logging
logger = logging.getLogger(__name__)

class DatabaseError(Exception):
pass

class DuplicateRecordError(DatabaseError):
pass

def with_retry(max_retries=3, delay=1):
def decorator(func):
   def wrapper(*args, **kwargs):
         for attempt in range(max_retries):
            try:
               return func(*args, **kwargs)
            except sqlite3.OperationalError as e:
               if attempt == max_retries - 1:
                     raise DatabaseError(f"操作失败，已重试{max_retries}次") from e
               logger.warning(f"第{attempt+1}次重试，错误: {e}")
               time.sleep(delay * (attempt + 1))# 指数退避
   return wrapper
return decorator

@with_retry(max_retries=3)
def create_user(phone, name):
try:
   with sqlite3.connect('app.db') as conn:
         conn.execute(
            "INSERT INTO users (phone, name) VALUES (?, ?)",
            (phone, name)
         )
         conn.commit()
         return {"id": conn.lastrowid, "phone": phone}
except sqlite3.IntegrityError as e:
   if "UNIQUE constraint" in str(e):
         raise DuplicateRecordError(f"手机号{phone}已存在") from e
   raise DatabaseError(f"数据完整性错误: {e}") from e

八、总结三条原则

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

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

热心网友5 发表于 2026-7-1 09:10:00

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

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

热心网友5 发表于 2026-7-1 09:10:00

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

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

热心网友5 发表于 2026-7-1 09:10:00

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

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

页: [1]

红客联盟 - 由08小组运营's Archiver