目录

  1. 引言：为何需要统计数据库更改？
    • 1.1 常见的统计场景
    • 1.2 Python sqlite3 模块概述
  2. 两个关键属性：cursor.rowcount 与 conn.total_changes
    • 2.1 理解数据库操作的“更改”
    • 2.2 Python sqlite3 的设计理念
  3. cursor.rowcount：单个操作的影响行数
    • 3.1 作用与特性
    • 3.2 INSERT, UPDATE, DELETE 操作的 rowcount
    • 3.3 SELECT 操作的 rowcount (特殊情况)
    • 3.4 Python 代码示例
  4. conn.total_changes：连接生命周期内的总更改数
    • 4.1 作用与特性
    • 4.2 累计更改的机制
    • 4.3 与 cursor.rowcount 的区别
    • 4.4 Python 代码示例
  5. 实际应用场景与选择
    • 5.1 何时使用 cursor.rowcount
    • 5.2 何时使用 conn.total_changes
  6. 最佳实践与注意事项
    • 6.1 事务管理 (COMMIT 与 ROLLBACK)
    • 6.2 使用 with 语句管理连接
    • 6.3 错误处理 (try-except)
    • 6.4 SELECT 语句后 rowcount 的局限性
  7. 综合代码示例
  8. 总结

1. 引言：为何需要统计数据库更改？

在与数据库交互的应用程序中，了解数据库操作的影响范围至关重要。统计更改行数可以帮助我们：

  • 验证操作：确认 INSERT、UPDATE 或 DELETE 语句是否按预期影响了正确数量的行。
  • 用户反馈：向用户提供清晰的反馈，例如“5条记录已成功更新”。
  • 审计与日志：记录哪些操作修改了多少数据，对系统行为进行追踪和审计。
  • 业务逻辑控制：基于更改数量执行后续的业务逻辑，例如，如果没有更新任何记录，则触发特定的错误处理或提醒。

1.2 Python sqlite3 模块概述

Python 的 sqlite3 模块是标准库的一部分，提供了与 SQLite 数据库交互的接口。它遵循 Python DB-API 2.0 规范，提供了一致的数据库操作方法。

2. 两个关键属性：cursor.rowcount 与 conn.total_changes

sqlite3 模块提供了两个核心属性来帮助我们统计数据库更改：

  1. cursor.rowcount：属于 Cursor 对象，表示最近一次 execute() 调用所影响的行数。
  2. conn.total_changes：属于 Connection 对象，表示自数据库连接打开以来，所有操作（通过该连接）对数据库的总更改行数。

2.1 理解数据库操作的“更改”

在数据库语境中，“更改”通常指 INSERT (插入)、UPDATE (更新) 和 DELETE (删除) 这三种数据修改语言 (DML) 操作。SELECT (查询) 操作虽然读取数据，但不会改变数据库中的数据，因此不计入“更改”数量。

2.2 Python sqlite3 的设计理念

sqlite3 模块旨在提供精确的反馈。cursor.rowcount 提供即时、特定操作的反馈，而 conn.total_changes 提供连接级别的、累积性的反馈。

3. cursor.rowcount：单个操作的影响行数

3.1 作用与特性

cursor.rowcount 是一个整数，表示上一个 cursor.execute() 调用（执行 INSERT, UPDATE, DELETE 或 REPLACE 语句）所影响的行数。

  • 瞬时性：它只反映紧邻其前的 execute() 操作的结果。
  • 局部性：它属于特定的 cursor 对象。如果你有多个游标，每个游标的 rowcount 都只反映该游标自身的最后一次操作。

3.2 INSERT, UPDATE, DELETE 操作的 rowcount

  • INSERT: 成功插入的行数 (通常是 1)。
  • UPDATE: 实际更新的行数。如果 WHERE 子句匹配了 0 行，则 rowcount 为 0。
  • DELETE: 实际删除的行数。如果 WHERE 子句匹配了 0 行，则 rowcount 为 0。

3.3 SELECT 操作的 rowcount (特殊情况)

对于 SELECT 语句，Python DB-API 2.0 规范要求 rowcount 属性在执行 SELECT 语句后应为 -1。这表示驱动程序无法确定受影响的行数（因为 SELECT 不影响行，它只是读取）。

3.4 Python 代码示例

import sqlite3
import os

DB_FILE = "rowcount_example.db"

def setup_db_for_rowcount():
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
        conn.commit()
    print(f"Database '{DB_FILE}' created for rowcount example.")

def demonstrate_rowcount():
    setup_db_for_rowcount()
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()

        # --- INSERT 操作 ---
        print("\n--- INSERT Operations ---")
        cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ('Alice', 30))
        print(f"INSERT rowcount: {cursor.rowcount}") # Expected: 1
        cursor.execute("INSERT INTO users (name, age) VALUES ('Bob', 25)")
        print(f"INSERT rowcount: {cursor.rowcount}") # Expected: 1
        
        # executemany 插入多行
        users_to_insert = [('Charlie', 35), ('David', 40), ('Eve', 28)]
        cursor.executemany("INSERT INTO users (name, age) VALUES (?, ?)", users_to_insert)
        print(f"INSERT (executemany) rowcount: {cursor.rowcount}") # Expected: 3

        # --- UPDATE 操作 ---
        print("\n--- UPDATE Operations ---")
        cursor.execute("UPDATE users SET age = ? WHERE name = ?", (31, 'Alice'))
        print(f"UPDATE (1 row) rowcount: {cursor.rowcount}") # Expected: 1
        cursor.execute("UPDATE users SET age = age + 1 WHERE age < 30")
        print(f"UPDATE (multiple rows) rowcount: {cursor.rowcount}") # Expected: 2 (Bob, Eve)
        cursor.execute("UPDATE users SET name = 'Unknown' WHERE id = 999") # No matching rows
        print(f"UPDATE (0 rows) rowcount: {cursor.rowcount}") # Expected: 0

        # --- DELETE 操作 ---
        print("\n--- DELETE Operations ---")
        cursor.execute("DELETE FROM users WHERE name = 'Charlie'")
        print(f"DELETE (1 row) rowcount: {cursor.rowcount}") # Expected: 1
        cursor.execute("DELETE FROM users WHERE age > 30")
        print(f"DELETE (multiple rows) rowcount: {cursor.rowcount}") # Expected: 2 (Alice, David)
        cursor.execute("DELETE FROM users WHERE name = 'NonExistent'") # No matching rows
        print(f"DELETE (0 rows) rowcount: {cursor.rowcount}") # Expected: 0
        
        # --- SELECT 操作 ---
        print("\n--- SELECT Operations ---")
        cursor.execute("SELECT * FROM users")
        print(f"SELECT rowcount: {cursor.rowcount}") # Expected: -1 (as per DB-API spec)

        conn.commit() # Commit all changes
        print("\nChanges committed.")

# demonstrate_rowcount() # Call this function to run the example

4. conn.total_changes：连接生命周期内的总更改数

4.1 作用与特性

conn.total_changes 是一个整数，表示自当前数据库连接 (conn) 打开以来，通过该连接执行的所有 INSERT, UPDATE, DELETE 操作影响的总行数。

  • 累积性：它会累加所有通过该连接执行的 DML 操作所影响的行数，而不管是在哪个游标上执行的。
  • 连接级别：它反映的是整个连接的活动。
  • 非事务性提交：total_changes 会立即反映更改，即使这些更改尚未通过 conn.commit() 持久化到数据库文件。然而，如果事务被 conn.rollback()，total_changes 的值不会回滚，它仍然会报告自连接打开以来的所有尝试的更改（这在某些DB-API实现中可能有所不同，但SQLite的total_changes通常是累加操作数）。

4.2 累计更改的机制

每次执行 INSERT, UPDATE, DELETE 语句时，相应的 cursor.rowcount 值会加到 conn.total_changes 上。

4.3 与 cursor.rowcount 的区别

特性 cursor.rowcount conn.total_changes
作用范围 单个 execute() 操作，特定 Cursor 对象 当前 Connection 对象的所有 DML 操作的累积和
时效性 仅反映最近一次 execute() 的结果 反映连接建立以来的所有 DML 操作的总和
类型 整数，对于 SELECT 通常为 -1 整数
用途 验证单个操作，提供即时反馈 审计整个会话，跟踪连接级别的总修改量

4.4 Python 代码示例

import sqlite3
import os

DB_FILE = "total_changes_example.db"

def setup_db_for_total_changes():
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT, stock INTEGER)")
        conn.commit()
    print(f"Database '{DB_FILE}' created for total_changes example.")

def demonstrate_total_changes():
    setup_db_for_total_changes()
    with sqlite3.connect(DB_FILE) as conn:
        cursor1 = conn.cursor()
        cursor2 = conn.cursor() # 使用第二个游标来证明 total_changes 是连接级别的

        print(f"\nInitial conn.total_changes: {conn.total_changes}") # Expected: 0

        # --- 游标1进行操作 ---
        cursor1.execute("INSERT INTO products (name, stock) VALUES (?, ?)", ('Laptop', 10))
        print(f"Cursor1 INSERT rowcount: {cursor1.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: 1, 1
        
        cursor1.execute("INSERT INTO products (name, stock) VALUES ('Mouse', 50)")
        print(f"Cursor1 INSERT rowcount: {cursor1.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: 1, 2

        # --- 游标2进行操作 ---
        cursor2.execute("UPDATE products SET stock = ? WHERE name = ?", (12, 'Laptop'))
        print(f"Cursor2 UPDATE rowcount: {cursor2.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: 1, 3 (注意这里是累计的)

        products_to_add = [('Keyboard', 20), ('Monitor', 5)]
        cursor2.executemany("INSERT INTO products (name, stock) VALUES (?, ?)", products_to_add)
        print(f"Cursor2 INSERT (executemany) rowcount: {cursor2.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: 2, 5

        cursor1.execute("DELETE FROM products WHERE stock < 10")
        print(f"Cursor1 DELETE rowcount: {cursor1.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: 1 (Monitor), 6

        # SELECT 不会增加 total_changes
        cursor1.execute("SELECT * FROM products")
        print(f"Cursor1 SELECT rowcount: {cursor1.rowcount}, conn.total_changes: {conn.total_changes}") # Expected: -1, 6

        conn.commit()
        print(f"\nAfter commit, final conn.total_changes: {conn.total_changes}") # Expected: 6

# demonstrate_total_changes() # Call this function to run the example

5. 实际应用场景与选择

5.1 何时使用 cursor.rowcount

  • 即时操作验证：在执行完一个具体的 INSERT, UPDATE, DELETE 语句后，立即检查是否成功影响了预期的行数。 
    cursor.execute("UPDATE users SET status = 'active' WHERE id = ?", (user_id,))
if cursor.rowcount == 1:
    print("User status updated successfully.")
else:
    print("Failed to update user status (user not found or no change).")
  • 批量操作统计：对于 executemany()，它会返回受影响的总行数。
  • 提供用户界面反馈：告知用户有多少条记录被修改。

5.2 何时使用 conn.total_changes

  • 会话级审计：在应用程序的整个数据库会话结束时，总结总共对数据库进行了多少次修改。
  • 资源管理：在某些情况下，如果 total_changes 达到某个阈值，可能需要触发自动备份或清理任务。
  • 检查整体活动：在程序某个阶段结束后，快速检查是否有数据被更改。

6. 最佳实践与注意事项

6.1 事务管理 (COMMIT 与 ROLLBACK)

  • cursor.rowcount 和 conn.total_changes 会立即反映操作的影响，无论是否 COMMIT。
  • 然而，只有 conn.commit() 才能将更改永久保存到数据库文件。如果没有 commit()，数据库关闭时所有未提交的更改都将丢失。
  • 如果执行 conn.rollback()，数据库会恢复到事务开始前的状态，但 conn.total_changes 的值不会回滚，它仍会反映自连接建立以来所有尝试的更改。这一点在实际应用中需要注意，total_changes 是一个“计数器”，而不是事务状态的反映。

6.2 使用 with 语句管理连接

sqlite3.connect() 对象是一个上下文管理器。使用 with 语句可以确保数据库连接在操作完成后自动关闭，即使发生错误也能正确释放资源。这也有助于在每个会话开始时重置 conn.total_changes。

6.3 错误处理 (try-except)

始终使用 try...except sqlite3.Error 来捕获可能发生的数据库错误，并进行适当的处理，而不是让程序崩溃。

6.4 SELECT 语句后 rowcount 的局限性

记住对于 SELECT 语句，cursor.rowcount 值为 -1。如果你需要知道 SELECT 查询返回了多少行，应该在获取结果后使用 len(cursor.fetchall())。

7. 综合代码示例

import sqlite3
import os
import time

DB_FILE = "full_changes_tracking.db"

def initialize_database():
    """初始化数据库，创建用户表并填充一些初始数据。"""
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT UNIQUE,
                status TEXT DEFAULT 'pending'
            );
        ''')
        # 初始插入一些用户
        initial_users = [
            ('Alice', 'alice@example.com', 'active'),
            ('Bob', 'bob@example.com', 'pending'),
            ('Charlie', 'charlie@example.com', 'active')
        ]
        cursor.executemany("INSERT INTO users (name, email, status) VALUES (?, ?, ?);", initial_users)
        conn.commit()
    print(f"Database '{DB_FILE}' initialized with 3 initial users.")

def display_all_users():
    """显示数据库中所有用户的信息。"""
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id, name, email, status FROM users ORDER BY id;")
        rows = cursor.fetchall()
        print("\n--- Current Users in Database ---")
        if not rows:
            print("  No users found.")
        else:
            for row in rows:
                print(f"  ID: {row[0]}, Name: {row[1]}, Email: {row[2]}, Status: {row[3]}")
        print("-" * 30)

def main():
    initialize_database()
    display_all_users()

    # 连接到数据库，开始跟踪更改
    print(f"\nConnecting to database: {DB_FILE}")
    try:
        with sqlite3.connect(DB_FILE) as conn:
            cursor = conn.cursor()
            print(f"Initial conn.total_changes: {conn.total_changes}") # 应该为0，因为连接是新的

            # --- 操作 1: 插入一个新用户 ---
            print("\n--- Operation 1: Inserting a new user ---")
            cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ('David', 'david@example.com'))
            print(f"  Last operation (INSERT) rowcount: {cursor.rowcount}")
            print(f"  Cumulative conn.total_changes: {conn.total_changes}") # 应该为1
            display_all_users() # 显示变更，但尚未提交

            # --- 操作 2: 更新多个用户的状态 ---
            print("\n--- Operation 2: Updating multiple users' status ---")
            cursor.execute("UPDATE users SET status = 'active' WHERE status = 'pending'")
            print(f"  Last operation (UPDATE) rowcount: {cursor.rowcount}")
            print(f"  Cumulative conn.total_changes: {conn.total_changes}") # 应该为1 + 2 = 3
            display_all_users()

            # --- 操作 3: 尝试更新一个不存在的用户 ---
            print("\n--- Operation 3: Updating a non-existent user ---")
            cursor.execute("UPDATE users SET status = 'inactive' WHERE id = 99")
            print(f"  Last operation (UPDATE) rowcount: {cursor.rowcount}") # 应该为0
            print(f"  Cumulative conn.total_changes: {conn.total_changes}") # 仍然是3 (因为没有实际更改)
            display_all_users()

            # --- 操作 4: 删除一个用户 ---
            print("\n--- Operation 4: Deleting a user ---")
            cursor.execute("DELETE FROM users WHERE name = 'Charlie'")
            print(f"  Last operation (DELETE) rowcount: {cursor.rowcount}") # 应该为1
            print(f"  Cumulative conn.total_changes: {conn.total_changes}") # 应该为3 + 1 = 4
            display_all_users()

            # --- 操作 5: 执行一个 SELECT 查询 ---
            print("\n--- Operation 5: Performing a SELECT query ---")
            cursor.execute("SELECT * FROM users WHERE status = 'active'")
            active_users = cursor.fetchall()
            print(f"  Last operation (SELECT) rowcount: {cursor.rowcount}") # 应该为-1
            print(f"  Number of active users found: {len(active_users)}")
            print(f"  Cumulative conn.total_changes: {conn.total_changes}") # 仍然是4 (SELECT不改变数据)
            
            # 提交所有更改
            conn.commit()
            print(f"\nAll changes committed to database.")
            print(f"Final conn.total_changes AFTER COMMIT: {conn.total_changes}") # 仍然是4

        print("\nDatabase connection closed.")
        # 在新的连接中查看最终状态
        display_all_users()

    except sqlite3.Error as e:
        print(f"An SQLite error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    main()

8. 总结

为您详尽解析了在 Python 中使用 sqlite3 模块来统计数据库更改行数的方法。

核心要点回顾:

  • cursor.rowcount: 针对单个 execute() 操作，反映 INSERT, UPDATE, DELETE 语句影响的行数。对于 SELECT 语句，其值为 -1。
  • conn.total_changes: 针对整个数据库连接，累积计算自连接打开以来所有 INSERT, UPDATE, DELETE 操作影响的总行数。
  • 应用场景: rowcount 适用于即时操作验证和用户反馈；total_changes 适用于会话级审计和整体活动跟踪。
  • 最佳实践: 始终使用参数化查询、with 语句管理连接、进行错误处理，并理解事务 COMMIT 和 ROLLBACK 对这两个属性的影响。

通过理解和正确使用 cursor.rowcount 和 conn.total_changes，您将能够更有效地验证数据库操作，提供更好的用户体验，并对应用程序中的数据更改进行全面的追踪。

# 数据库 # python # sqlite
Logo
亚马逊云科技技术品牌专区

更多推荐

两分钟安装 Hermes Agent

在当今快速发展的科技时代，人工智能（artificial intelligence，AI）技术正以惊人的速度改变着人们的生活和工作方式。在这个新时代的浪潮中，大模型技术成为AI领域的一颗耀眼新星。ChatGPT作为大模型技术的重要应用之一，正在引领着人机交互领域的革新浪潮。本书将带领读者深入探索大模型新时代，通过ChatGPT实战项目和内部解析，深入掌握基于ChatGPT的大模型应用开发领域的关键

avatar 亚马逊云科技技术品牌专区

【人工智能】AI基础知识（第七节：RAG）

RAG（检索增强生成）技术解决了大模型在处理私有和实时数据时的局限性。其核心流程包括：1）将用户提问向量化，通过语义搜索匹配相似内容；2）检索向量数据库找出最相关的TopK数据片段；3）将检索结果与原始问题结合形成增强提示；4）由大模型生成最终答案。通过向量化技术（将词语特征转化为多维数值）和余弦相似度计算，RAG能有效处理语义相近的查询（如"退商品"与"退货"），从而避免大模型的"幻觉"问题，为

avatar 亚马逊云科技技术品牌专区
cover

大白话讲透RAG！彻底搞懂检索增强生成原理

avatar 亚马逊云科技技术品牌专区