免费编程软件「python+pycharm」
链接：https://pan.quark.cn/s/48a86be2fdc0
凌晨两点的报警电话
去年冬天的一个凌晨，我被手机吵醒了。监控报警：用户注册接口成功率跌到了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管理，避免资源泄漏。 - 把异常包装成业务异常：让上层代码按业务逻辑处理，而不是处理数据库底层错误。
那次凌晨的报警之后，我把所有数据库操作都按照这个模板重构了。后来再遇到网络抖动、重复提交这类问题，代码都能正确处理——要么重试成功，要么返回明确的业务提示，再也没有半夜的报警电话了。
