Python中数据库操作与异常处理详解

凌晨两点的报警电话

去年冬天的一个凌晨，我被手机吵醒了。监控报警：用户注册接口成功率跌到了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

总结：记住三条原则

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

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