Python sqlite3.IntegrityError 异常

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

这份全面的指南探讨了 Python 的 sqlite3.IntegrityError 异常，该异常在数据库约束被违反时发生。我们将介绍常见原因、处理技巧和实际示例。

基本定义

sqlite3.IntegrityError 是 sqlite3.DatabaseError 的一个子类，表示 SQLite 数据库中的约束违反。当操作会破坏数据库完整性规则时，它会被引发。

常见的触发因素包括：主键冲突、外键冲突、NOT NULL 约束失败以及 UNIQUE 约束冲突。正确处理可确保数据一致性和应用程序的健壮性。

主键冲突

此示例演示了插入重复 ID 时的主键冲突。

primary_key.py

import sqlite3

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS products
                          (id INTEGER PRIMARY KEY, name TEXT)''')
        
        # First insert succeeds
        cursor.execute("INSERT INTO products VALUES (1, 'Laptop')")
        
        # Second insert with same ID raises IntegrityError
        cursor.execute("INSERT INTO products VALUES (1, 'Phone')")
        
except sqlite3.IntegrityError as e:
    print(f"Integrity error occurred: {e}")

第二个插入失败，因为它尝试重用主键值 1。主键在一个表中必须是唯一的。错误消息指示了具体的约束冲突。

此示例使用上下文管理器来确保正确清理资源，即使发生异常也是如此。事务会在出错时自动回滚。

NOT NULL 约束冲突

此示例显示了尝试将 NULL 插入到 NOT NULL 列中时会发生什么。

not_null.py

import sqlite3

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                          (id INTEGER PRIMARY KEY,
                           name TEXT NOT NULL,
                           department TEXT NOT NULL)''')
        
        # Valid insert
        cursor.execute("INSERT INTO employees (name, department) VALUES (?, ?)",
                      ('Alice', 'HR'))
        
        # Invalid insert - missing department
        cursor.execute("INSERT INTO employees (name) VALUES (?)", ('Bob',))
        
except sqlite3.IntegrityError as e:
    print(f"NOT NULL constraint failed: {e}")

第二个插入失败，因为它没有为标记为 NOT NULL 的 'department' 列提供值。 SQLite 拒绝该操作以维护数据完整性。

错误消息明确指出违反了哪个列约束。这有助于调试和修复应用程序代码中的问题。

UNIQUE 约束冲突

此示例演示了插入重复值时的 UNIQUE 约束冲突。

unique.py

import sqlite3

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS users
                          (id INTEGER PRIMARY KEY,
                           email TEXT UNIQUE,
                           username TEXT UNIQUE)''')
        
        # First insert succeeds
        cursor.execute("INSERT INTO users (email, username) VALUES (?, ?)",
                      ('alice@example.com', 'alice'))
        
        # Second insert with duplicate email raises IntegrityError
        cursor.execute("INSERT INTO users (email, username) VALUES (?, ?)",
                      ('alice@example.com', 'alice2'))
        
except sqlite3.IntegrityError as e:
    print(f"UNIQUE constraint failed: {e}")

UNIQUE 约束确保没有两行在指定的列中具有相同的值。在这里，尝试插入重复的电子邮件地址会触发错误。

该示例展示了如何正确使用参数化查询来防止 SQL 注入，同时演示约束冲突。

外键冲突

此示例说明了当引用的行不存在时的外键冲突。

foreign_key.py

import sqlite3

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        
        # Enable foreign key constraints (SQLite has them off by default)
        cursor.execute("PRAGMA foreign_keys = ON")
        
        cursor.execute('''CREATE TABLE IF NOT EXISTS departments
                          (id INTEGER PRIMARY KEY, name TEXT)''')
        
        cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                          (id INTEGER PRIMARY KEY,
                           name TEXT,
                           dept_id INTEGER,
                           FOREIGN KEY(dept_id) REFERENCES departments(id))''')
        
        # Insert into employees without corresponding department
        cursor.execute("INSERT INTO employees (name, dept_id) VALUES (?, ?)",
                      ('Bob', 99))
        
except sqlite3.IntegrityError as e:
    print(f"Foreign key constraint failed: {e}")

插入失败，因为部门 ID 99 在 departments 表中不存在。外键强制执行相关表之间的引用完整性。

请注意，SQLite 需要使用 PRAGMA 语句显式启用外键支持。这是外键不起作用时的常见困惑来源。

CHECK 约束冲突

此示例显示了当值不满足条件时的 CHECK 约束冲突。

check.py

import sqlite3

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS products
                          (id INTEGER PRIMARY KEY,
                           name TEXT,
                           price REAL CHECK(price > 0),
                           stock INTEGER CHECK(stock >= 0))''')
        
        # Valid insert
        cursor.execute("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
                      ('Widget', 9.99, 100))
        
        # Invalid insert - negative price
        cursor.execute("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
                      ('Gadget', -1.99, 50))
        
except sqlite3.IntegrityError as e:
    print(f"CHECK constraint failed: {e}")

CHECK 约束强制执行 price 必须大于 0。第二个插入尝试设置负价格，这违反了此规则。

CHECK 约束非常强大，可以在数据库级别强制执行业务规则，确保无论应用程序逻辑如何都无法存储无效数据。

处理多个约束

此示例展示了如何违反和一起处理多个约束。

multiple.py

import sqlite3

def handle_integrity_error(error):
    if "NOT NULL" in str(error):
        print("Error: Missing required field")
    elif "UNIQUE" in str(error):
        print("Error: Duplicate value")
    elif "FOREIGN KEY" in str(error):
        print("Error: Invalid reference")
    else:
        print(f"Database integrity error: {error}")

try:
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS orders
                          (id INTEGER PRIMARY KEY,
                           customer_id INTEGER NOT NULL,
                           product_id INTEGER NOT NULL,
                           quantity INTEGER CHECK(quantity > 0),
                           FOREIGN KEY(customer_id) REFERENCES customers(id),
                           FOREIGN KEY(product_id) REFERENCES products(id))''')
        
        # This will fail multiple constraints
        cursor.execute("INSERT INTO orders (customer_id) VALUES (NULL)")
        
except sqlite3.IntegrityError as e:
    handle_integrity_error(e)

此示例演示了一个复杂的错误处理程序，该处理程序检查错误消息以确定违反了哪个约束。插入失败，同时违反了 NOT NULL 和 FOREIGN KEY 约束。

该处理程序根据具体的约束失败提供用户友好的消息，从而改善应用程序中的用户体验。

从 IntegrityError 中恢复

此示例显示了如何通过使用有效数据重试从 IntegrityError 中恢复。

recovery.py

import sqlite3

def create_tables(conn):
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS users
                      (id INTEGER PRIMARY KEY,
                       username TEXT UNIQUE)''')

def add_user(conn, username):
    try:
        cursor = conn.cursor()
        cursor.execute("INSERT INTO users (username) VALUES (?)", (username,))
        conn.commit()
        return True
    except sqlite3.IntegrityError:
        print(f"Username '{username}' already exists")
        return False

with sqlite3.connect('users.db') as conn:
    create_tables(conn)
    
    # First attempt succeeds
    add_user(conn, 'alice')
    
    # Second attempt fails
    if not add_user(conn, 'alice'):
        # Recovery: try alternative username
        add_user(conn, 'alice2')
    
    # Verify results
    cursor = conn.cursor()
    cursor.execute("SELECT username FROM users")
    print("Current users:", [row[0] for row in cursor.fetchall()])

当发生重复的用户名错误时，代码会通过尝试备用用户名来优雅地恢复。这演示了实际应用程序中强大的错误处理能力。

该示例将数据库操作分离到干净的函数中，并展示了如何在处理完整性约束的同时构建代码以实现可维护性。

最佳实践

首先验证数据： 在插入之前，在应用程序代码中检查约束
使用事务： 将相关操作分组以保持一致性
提供清晰的消息： 解析错误消息以获得用户友好的反馈
考虑约束： 使用适当的约束设计数据库架构
优雅处理： 为常见的约束冲突规划恢复路径

资料来源

作者

我的名字是 Jan Bodnar，我是一位热情的程序员，拥有丰富的编程经验。我从 2007 年开始撰写编程文章。至今，我已撰写了 1,400 多篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 Python 教程。