ZetCode

Python sqlite3.DatabaseError 异常

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

本综合指南探讨 Python 的 sqlite3.DatabaseError 异常，它处理 SQLite 操作中与数据库相关的错误。

基本定义

sqlite3.DatabaseError 是 Python 的 sqlite3 模块中与数据库相关的异常的基类。 它继承自 sqlite3.Error。

此异常针对与数据库本身相关的错误而引发，而不是编程错误。 它涵盖诸如数据库损坏、磁盘 I/O 错误以及其他特定于数据库的问题。

处理 DatabaseError

此示例显示了使用 SQLite 数据库时 DatabaseError 的基本错误处理。

basic_error_handling.py 
import sqlite3

try:
    with sqlite3.connect("example.db") as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM non_existent_table")
except sqlite3.DatabaseError as e:
    print(f"Database error occurred: {e}")
except Exception as e:
    print(f"Other error occurred: {e}")

此代码尝试查询一个不存在的表，这将引发 DatabaseError。 with 语句确保资源被正确关闭。

该示例演示了正确的异常处理层次结构，首先捕获特定的数据库错误，然后再捕获更一般的异常。

损坏的数据库文件

此示例显示了当使用损坏的 SQLite 数据库文件时如何引发 DatabaseError。

corrupt_database.py 
import sqlite3

try:
    with sqlite3.connect("corrupt.db") as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM test")
except sqlite3.DatabaseError as e:
    print(f"Database corruption detected: {e}")
    # Here you might attempt recovery or notify the user

当 SQLite 检测到数据库损坏时，它会引发 DatabaseError。 确切的消息取决于损坏的类型和位置。

在生产代码中，你可以在检测到损坏时实施恢复程序或用户通知。

磁盘 I/O 错误

此示例演示了处理显示为 DatabaseError 异常的磁盘 I/O 错误。

disk_io_error.py 
import sqlite3

try:
    # Attempt to create database in read-only location
    with sqlite3.connect("/readonly/location/test.db") as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE test (id INTEGER)")
except sqlite3.DatabaseError as e:
    print(f"Disk I/O error occurred: {e}")
    # Handle the error (e.g., fallback location)

当 SQLite 无法读取或写入数据库文件时，会发生磁盘 I/O 错误。 这可能是由于权限或文件系统问题引起的。

该示例演示了如何捕获这些错误，并可能在你的应用程序中实施回退行为。

事务错误

此示例显示了与事务相关的问题如何引发 DatabaseError 异常。

transaction_error.py 
import sqlite3

try:
    with sqlite3.connect("transactions.db") as conn:
        cursor = conn.cursor()
        cursor.execute("BEGIN IMMEDIATE")
        cursor.execute("INSERT INTO accounts VALUES (1, 100)")
        # Simulate error during transaction
        raise RuntimeError("Unexpected error")
        conn.commit()
except sqlite3.DatabaseError as e:
    print(f"Transaction error: {e}")
except Exception as e:
    print(f"Other error: {e}")
    # Connection's __exit__ will rollback automatically

提交或回滚更改时出现问题，可能会发生事务错误。 with 语句确保在异常发生时正确回滚。

当事务期间发生错误时，这种模式对于维护数据库一致性至关重要。

数据库模式错误

此示例演示了处理与模式相关的 DatabaseError 异常。

schema_error.py 
import sqlite3

try:
    with sqlite3.connect("schema.db") as conn:
        cursor = conn.cursor()
        # Attempt invalid schema change
        cursor.execute("ALTER TABLE users ADD COLUMN email UNIQUE")
except sqlite3.DatabaseError as e:
    print(f"Schema modification failed: {e}")
    # Handle the error (e.g., alternative schema change)

当尝试无效的数据库模式修改时，会发生模式错误。 SQLite 对必须遵循的模式更改有特定规则。

该示例演示了如何捕获这些错误，并可能实施替代的模式修改策略。

数据库锁定错误

此示例显示了如何处理引发 DatabaseError 的数据库锁定问题。

locking_error.py 
import sqlite3
import time

def attempt_operation():
    try:
        with sqlite3.connect("busy.db", timeout=1.0) as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE counters SET value = value + 1")
            conn.commit()
            return True
    except sqlite3.DatabaseError as e:
        print(f"Database locked, retrying... ({e})")
        return False

# Retry logic for locked database
for attempt in range(3):
    if attempt_operation():
        break
    time.sleep(1)
else:
    print("Failed to acquire database lock after 3 attempts")

当多个进程尝试同时修改数据库时，会发生锁定错误。 该示例实现了一个重试机制。

timeout 参数指定在引发异常之前等待锁定的时间。

自定义错误处理

此示例演示了高级错误处理，针对不同的 DatabaseError 场景具有自定义行为。

custom_error_handling.py 
import sqlite3

def handle_database_error(error):
    error_msg = str(error).lower()
    if "corrupt" in error_msg:
        print("Database corruption detected - initiating recovery")
        # Start recovery process
    elif "locked" in error_msg:
        print("Database locked - implement retry logic")
        # Implement retry logic
    elif "disk i/o" in error_msg:
        print("Disk I/O error - check storage")
        # Check storage availability
    else:
        print(f"Unhandled database error: {error}")

try:
    with sqlite3.connect("app.db") as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM critical_data")
except sqlite3.DatabaseError as e:
    handle_database_error(e)
except Exception as e:
    print(f"Non-database error occurred: {e}")

此示例显示了如何实施复杂的错误处理，以不同的方式响应各种数据库错误情况。

自定义处理程序检查错误消息，以确定适用于不同场景的适当恢复操作。

最佳实践

资料来源

作者

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

列出所有 Python 教程。