python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python数据库操作

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

作者：站大爷IP

这篇文章主要介绍了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。它表示你的代码写错了：

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

数据完整性层面的异常

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

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

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

业务场景里的正确姿势

回到注册接口的案例。当捕获到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. 把异常包装成业务异常：让上层代码按业务逻辑处理，而不是处理数据库底层错误。

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

到此这篇关于Python中数据库操作与异常处理详解的文章就介绍到这了,更多相关Python数据库操作内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文