先讲个真实的事。

去年冬天凌晨两点，手机响了。监控报警：用户注册接口成功率跌到50%。

从被窝里爬起来翻日志，发现全是同一个异常：

IntegrityError: (1062, "Duplicate entry '13800138000' for key 'phone'")

这问题其实挺典型的。一个用户注册功能，用户填完手机号提交，系统就往数据库里插一条新记录。正常情况下手机号是唯一的，这没问题。但那天网络不好，用户点了两次“提交”，后端接口收到了两个一模一样的注册请求。

第一个请求正常跑了，插入了数据。第二个请求来了之后，发现手机号已经存在，INSERT语句直接爆了。然后代码做了什么？把异常打印到了日志里，给前端返回了一个“服务器错误”。

用户那边看到的是报错页面，但他不知道的是——第一次提交其实已经成功了。他以为注册失败，可能又试了第三次、第四次……

这个Bug暴露了两个问题：接口没有做幂等处理，数据库操作的异常处理写得太粗糙。

今天就从这里出发，把Python操作数据库时应该怎么处理异常这件事，掰开揉碎了说清楚。

数据库操作到底有哪些异常？

在Python里操作数据库，不管是SQLite、MySQL还是PostgreSQL，常见的异常可以归成几类。

连接层面的异常

import sqlite3

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

OperationalError是操作层面的错误，跟SQL语句本身没关系，是执行环境出了问题。比如：

• 数据库服务器挂了
• 连接超时
• 文件被锁
• 磁盘满了

这类错误通常需要重试，或者走降级方案。

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

# 表不存在
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在SQLite里是OperationalError的子集，在MySQL的pymysql里有独立的ProgrammingError。它表示代码写错了：

• 表名或列名拼写错误
• SQL语法有问题
• 参数绑定的数量不匹配

这类错误不应该用重试来解决，正确的做法是修代码。

数据完整性层面的异常

# 违反了唯一约束
cursor.execute("INSERT INTO users (phone) VALUES ('13800138000')")
# sqlite3.IntegrityError: UNIQUE constraint failed: users.phone

IntegrityError是数据完整性错误，常见于：

• 主键冲突（重复插入同一主键）
• 唯一键冲突（重复的手机号、邮箱）
• 外键约束失败（引用了不存在的父表记录）
• NOT NULL约束失败

这类异常往往需要业务逻辑来处理——不是简单报错就完事了，要判断该怎么做。

业务场景里的正确姿势

回到注册接口的案例。当捕获到IntegrityError时，正确的做法不是直接返回错误，而是区分情况：

import sqlite3

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把所有异常都吞掉，否则既不知道出了什么问题，也没法针对性地处理。

上下文管理器：让你的数据库代码自动“善后”

很多人的数据库代码长这样：

def get_user(user_id):
conn = sqlite3.connect('app.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
result = cursor.fetchone()
cursor.close()
conn.close()
return result

万一execute报错了怎么办？cursor.close()和conn.close()都不会被执行，连接就泄漏了。

用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会自动关闭，即使发生了异常

with块保证资源被正确释放。如果你用的是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

总结：记住三条原则

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

那次凌晨的报警之后，把所有数据库操作都按照这个模板重构了。后来再遇到网络抖动、重复提交这类问题，代码都能正确处理——要么重试成功，要么返回明确的业务提示，再也没有半夜的报警电话了。