Python使用sqlite3时如何无错终止运行中的SQL执行

测试场景下不需要保留全量写入数据，核心要解决的是中断时没有做数据库连接收尾导致异常抛出的问题，以下是可直接落地的方案：

方案1：捕获KeyboardInterrupt异常做安全收尾 #

这是最易实现的方案，直接在代码外层包裹异常捕获逻辑，触发中断时主动回滚未提交的事务、关闭数据库连接后正常退出，不会抛出错误堆栈。

import sqlite3
import sys

conn = None
try:
    conn = sqlite3.connect('test_data.db')
    cur = conn.cursor()
    # 下方替换为你的大批量数据写入逻辑
    for i in range(1000000):
        cur.execute("INSERT INTO test_table (num) VALUES (?)", (i,))
except KeyboardInterrupt:
    print("\n已触发终止，正在清理连接退出...")
    if conn:
        # 测试场景无需保存数据，直接回滚所有未提交操作
        conn.rollback()
        conn.close()
    sys.exit(0)

注意事项：

• 不要开启自动提交（即不要在connect()时传入isolation_level=None），默认配置下未提交的写入操作会暂存在事务缓存中，回滚不会损坏数据库文件
• 不要在循环中加conn.commit()，否则不仅写入速度慢，中断时还会残留部分已提交的测试数据

方案2：注册信号处理函数接管中断信号 #

如果中断刚好触发在sqlite3底层C逻辑执行阶段，外层的异常捕获可能会失效，用信号处理接管SIGINT（Ctrl+C对应的系统信号）稳定性更高：

import sqlite3
import signal
import sys

conn = None

def exit_handler(signum, frame):
    print("\n已触发终止，正在清理连接退出...")
    if conn:
        conn.rollback()
        conn.close()
    sys.exit(0)

# 注册Ctrl+C信号处理逻辑
signal.signal(signal.SIGINT, exit_handler)

conn = sqlite3.connect('test_data.db')
cur = conn.cursor()
# 下方替换为你的大批量数据写入逻辑
for i in range(1000000):
    cur.execute("INSERT INTO test_table (num) VALUES (?)", (i,))

测试场景专属优化 #

如果你完全不需要保留任何写入数据，连磁盘文件都不想生成，可以直接使用sqlite3的内存模式，所有操作都在内存中执行，测试运行速度更快，中断时也不会留下残留文件：

# 连接内存数据库，不会生成磁盘文件
conn = sqlite3.connect(':memory:')

内容的提问来源于stack exchange，提问作者mansour32