Python sqlite3.Error 异常

上次修改时间：2025 年 4 月 15 日

本综合指南探讨 Python 的 sqlite3.Error 异常，它是所有 SQLite 数据库错误的基本类。我们将涵盖错误处理、特定异常类型和实际示例。

基本定义

sqlite3.Error 是 SQLite 数据库错误的基本异常类。当 SQLite 在数据库操作期间遇到问题时，会引发此异常。

主要特征：它通过其消息提供错误详细信息，具有针对不同错误类型的特定子类，并通过发出问题信号来帮助维护数据库完整性。

基本错误处理

此示例演示了在执行 SQL 语句时使用 sqlite3.Error 的基本错误处理。

import sqlite3

try:
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        # This will raise sqlite3.Error if table doesn't exist
        cursor.execute("SELECT * FROM non_existent_table")
        print(cursor.fetchall())
except sqlite3.Error as e:
    print(f"Database error occurred: {e}")

此示例演示了捕获基本 sqlite3.Error 异常。 with 语句确保即使发生错误也能正确清理资源。

错误消息将指示出现什么问题，例如在本例中为“no such table”（没有这样的表）。 始终处理数据库错误以防止崩溃。

处理特定错误类型

SQLite 为不同的错误情况提供了特定的异常子类。 此示例演示了如何分别处理它们。

import sqlite3

try:
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        # Attempt to create table with invalid SQL
        cursor.execute("CREATE TABLE users (id INVALID_TYPE)")
except sqlite3.OperationalError as e:
    print(f"Operational error: {e}")
except sqlite3.ProgrammingError as e:
    print(f"Programming error: {e}")
except sqlite3.Error as e:
    print(f"Generic database error: {e}")

此示例显示了捕获特定 SQLite 异常。 OperationalError 用于 SQL 执行问题，而 ProgrammingError 用于 API 误用。

处理特定异常可以实现更精确的错误恢复和用户反馈。 基本 sqlite3.Error 捕获任何未处理的情况。

数据库完整性错误

此示例演示了如何处理 sqlite3.IntegrityError，该错误发生在违反数据库约束时。

import sqlite3

try:
    with sqlite3.connect(':memory:') as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT UNIQUE)")
        cursor.execute("INSERT INTO users (email) VALUES ('test@example.com')")
        # This will violate UNIQUE constraint
        cursor.execute("INSERT INTO users (email) VALUES ('test@example.com')")
except sqlite3.IntegrityError as e:
    print(f"Integrity constraint violated: {e}")
except sqlite3.Error as e:
    print(f"Other database error: {e}")

该示例创建一个具有 UNIQUE 约束的表，并尝试插入重复的电子邮件。 这会引发 IntegrityError。

完整性错误对于维护数据一致性至关重要。 应始终在应用程序中捕获并妥善处理它们。

数据库锁定错误

此示例显示了当数据库被另一个进程锁定时如何处理 sqlite3.OperationalError。

import sqlite3
import time

def process1():
    with sqlite3.connect('locked.db') as conn:
        cursor = conn.cursor()
        cursor.execute("BEGIN EXCLUSIVE")
        cursor.execute("UPDATE counters SET value = value + 1")
        time.sleep(5)  # Hold lock for 5 seconds
        conn.commit()

def process2():
    try:
        with sqlite3.connect('locked.db', timeout=1) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM counters")
            print(cursor.fetchall())
    except sqlite3.OperationalError as e:
        print(f"Database locked error: {e}")

# In real code, these would be separate processes
process1()
process2()

该示例模拟了两个进程访问同一个数据库。 当第二个进程无法获得锁时，它会失败并出现 OperationalError。

数据库锁定在多进程应用程序中很常见。 应实施适当的错误处理和重试逻辑以提高鲁棒性。

错误时事务回滚

此示例演示了当上下文管理器中发生错误时自动事务回滚。

import sqlite3

try:
    with sqlite3.connect('transactions.db') as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE IF NOT EXISTS accounts (id INTEGER, balance REAL)")
        cursor.execute("INSERT INTO accounts VALUES (1, 100.0)")
        cursor.execute("INSERT INTO accounts VALUES (2, 200.0)")
        
        # This will cause an error (division by zero)
        invalid = 1 / 0
        
        cursor.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
        cursor.execute("UPDATE accounts SET balance = balance + 50 WHERE id = 2")
except sqlite3.Error as e:
    print(f"Database error: {e}")
except ZeroDivisionError as e:
    print(f"Application error: {e}")
    # Check that transaction was rolled back
    with sqlite3.connect('transactions.db') as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM accounts")
        print("Accounts after error:", cursor.fetchall())

该示例显示上下文管理器中的任何异常（数据库或其他）都会触发回滚。 初始插入被还原。

当复杂操作期间发生错误时，此自动回滚行为有助于维护数据库一致性。

自定义错误处理函数

此示例演示了如何为数据库操作创建自定义错误处理程序。

import sqlite3
from typing import Callable

def db_operation(operation: Callable, error_handler: Callable):
    try:
        with sqlite3.connect('example.db') as conn:
            return operation(conn)
    except sqlite3.Error as e:
        return error_handler(e)

def query_users(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    return cursor.fetchall()

def handle_error(e):
    print(f"Custom error handling: {e}")
    return []

# Usage
result = db_operation(query_users, handle_error)
print("Query result:", result)

该示例将数据库操作包装在一个函数中，该函数以一致的方式处理错误。 with 语句确保正确清理。

此模式对于具有许多需要统一错误处理和资源管理的数据库调用的应用程序非常有用。

最佳实践

• 始终处理异常： 永远不要让数据库错误在未经检查的情况下传播
• 使用特定异常： 捕获最具体的异常
• 清理资源： 使用上下文管理器进行自动清理
• 提供用户反馈： 将技术错误翻译成用户友好的消息
• 记录错误： 记录详细的错误信息以进行调试

资料来源

• Python sqlite3 文档
• SQLite 结果代码
• SQLite 事务

作者

我的名字是 Jan Bodnar，我是一位充满热情的程序员，拥有丰富的编程经验。 我自 2007 年以来一直在撰写编程文章。 迄今为止，我已经撰写了超过 1,400 篇文章和 8 本电子书。 我拥有超过十年的编程教学经验。

列出所有 Python 教程。