Python sqlite3.Connection.set_trace_callback 方法
上次修改时间：2025 年 4 月 15 日
这篇综合指南探讨了 Python 的 sqlite3.Connection.set_trace_callback 方法，它允许追踪连接执行的 SQL 语句。我们将涵盖基本用法、实际示例和调试技巧。
基本定义
set_trace_callback 方法注册一个回调函数，该函数针对连接执行的每个 SQL 语句被调用。 它对于调试、日志记录或分析 SQL 查询很有用。
主要特点：回调函数接收 SQL 文本作为字符串，它在语句执行之前被调用，并且可以用于监视所有数据库操作。可以将回调函数设置为 None 以禁用跟踪。
基本跟踪回调示例
此示例演示如何设置一个简单的跟踪回调，该回调在执行之前打印所有 SQL 语句。
basic_trace.py
import sqlite3
def trace_callback(sql):
print(f"Executing SQL: {sql}")
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(trace_callback)
cursor = conn.cursor()
cursor.execute("CREATE TABLE users (id INTEGER, name TEXT)")
cursor.execute("INSERT INTO users VALUES (1, 'Alice')")
cursor.execute("SELECT * FROM users")
print(cursor.fetchall())
跟踪回调在执行之前打印每个 SQL 语句。 这有助于了解正在数据库上执行哪些操作。
输出将按顺序显示 CREATE TABLE、INSERT 和 SELECT 语句，然后是查询结果。 回调函数接收确切的 SQL 文本。
过滤特定 SQL 语句
此示例展示了如何使用跟踪回调来过滤和处理特定类型的 SQL 语句。
filter_statements.py
import sqlite3
def trace_callback(sql):
sql_lower = sql.lower()
if "insert" in sql_lower:
print(f"Insert operation detected: {sql}")
elif "select" in sql_lower:
print(f"Select operation detected: {sql}")
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(trace_callback)
cursor = conn.cursor()
cursor.execute("CREATE TABLE logs (id INTEGER, message TEXT)")
cursor.execute("INSERT INTO logs VALUES (1, 'Startup')")
cursor.execute("SELECT * FROM logs")
cursor.execute("DELETE FROM logs WHERE id = 1")
conn.commit()
回调函数现在检查特定的 SQL 关键字，并且仅打印包含这些关键字的语句。 这对于监视特定操作很有用。
请注意，回调函数仍然接收所有语句，但我们有选择地仅处理我们感兴趣的语句。 此处不会打印 DELETE 语句。
测量查询执行时间
此示例演示如何使用带有时间测量的跟踪回调来分析查询性能。
query_timing.py
import sqlite3
import time
class QueryTimer:
def __init__(self):
self.start_time = None
def trace_callback(self, sql):
self.start_time = time.time()
print(f"Starting query: {sql}")
def print_duration(self):
if self.start_time:
duration = (time.time() - self.start_time) * 1000
print(f"Query completed in {duration:.2f} ms\n")
timer = QueryTimer()
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(timer.trace_callback)
cursor = conn.cursor()
# Create and populate a table
cursor.execute("CREATE TABLE data (id INTEGER, value REAL)")
cursor.executemany("INSERT INTO data VALUES (?, ?)",
[(i, i * 1.5) for i in range(1000)])
timer.print_duration()
# Run a query
cursor.execute("SELECT AVG(value) FROM data")
timer.print_duration()
print("Average:", cursor.fetchone()[0])
此示例使用一个类来跟踪查询开始时间并测量持续时间。 跟踪回调记录查询开始的时间，我们测量之后的经过时间。
输出显示正在执行的 SQL 以及每个操作花费的时间。 这对于性能优化和识别慢查询很有价值。
统计数据库操作
此示例使用跟踪回调来统计会话期间执行的不同类型的数据库操作。
operation_counter.py
import sqlite3
class OperationCounter:
def __init__(self):
self.counts = {
'select': 0,
'insert': 0,
'update': 0,
'delete': 0,
'other': 0
}
def trace_callback(self, sql):
sql_lower = sql.lower()
if "select" in sql_lower:
self.counts['select'] += 1
elif "insert" in sql_lower:
self.counts['insert'] += 1
elif "update" in sql_lower:
self.counts['update'] += 1
elif "delete" in sql_lower:
self.counts['delete'] += 1
else:
self.counts['other'] += 1
def print_counts(self):
print("Database operation counts:")
for op, count in self.counts.items():
print(f"{op.capitalize()}: {count}")
counter = OperationCounter()
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(counter.trace_callback)
cursor = conn.cursor()
cursor.execute("CREATE TABLE products (id INTEGER, name TEXT, price REAL)")
cursor.executemany("INSERT INTO products VALUES (?, ?, ?)",
[(1, 'Apple', 1.2), (2, 'Banana', 0.8)])
cursor.execute("UPDATE products SET price = 1.3 WHERE name = 'Apple'")
cursor.execute("SELECT * FROM products")
cursor.execute("SELECT COUNT(*) FROM products")
cursor.execute("DELETE FROM products WHERE price > 1.0")
counter.print_counts()
OperationCounter 类维护不同 SQL 操作类型的计数。 跟踪回调为每个语句增加相应的计数器。
这对于了解数据库上的工作负载模式很有用，这有助于优化和资源规划。 最终计数在最后打印。
将 SQL 语句记录到文件
此示例展示了如何使用跟踪回调将所有 SQL 语句记录到文件，这对于审计或调试很有用。
sql_logger.py
import sqlite3
from datetime import datetime
class SQLLogger:
def __init__(self, filename):
self.log_file = open(filename, 'a')
def trace_callback(self, sql):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.log_file.write(f"[{timestamp}] {sql}\n")
self.log_file.flush()
def close(self):
self.log_file.close()
logger = SQLLogger("sql_log.txt")
try:
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(logger.trace_callback)
cursor = conn.cursor()
cursor.execute("CREATE TABLE events (id INTEGER, ts TEXT, desc TEXT)")
cursor.execute("INSERT INTO events VALUES (1, datetime('now'), 'Start')")
cursor.execute("SELECT * FROM events")
# Simulate an error
try:
cursor.execute("INVALID SQL STATEMENT")
except sqlite3.OperationalError:
pass
finally:
logger.close()
print("SQL statements logged to sql_log.txt")
SQLLogger 类将每个 SQL 语句以及时间戳写入文件。 这会创建所有数据库操作的审计跟踪，包括其时间安排。
日志文件对于调试问题非常有价值，尤其是在生产环境中，您需要了解问题发生之前发生了什么。 文件在 finally 块中正确关闭。
检测潜在的 SQL 注入
此示例演示如何通过在 SQL 语句中查找可疑模式，使用跟踪回调来检测潜在的 SQL 注入尝试。
injection_detector.py
import sqlite3
import re
class InjectionDetector:
def __init__(self):
self.suspicious_patterns = [
r";\s*--",
r";\s*#",
r"union\s+select",
r"1\s*=\s*1",
r"sleep\s*\("
]
def trace_callback(self, sql):
for pattern in self.suspicious_patterns:
if re.search(pattern, sql, re.IGNORECASE):
print(f"WARNING: Potential SQL injection detected!")
print(f"Suspicious SQL: {sql}")
break
detector = InjectionDetector()
with sqlite3.connect(":memory:") as conn:
conn.set_trace_callback(detector.trace_callback)
cursor = conn.cursor()
# Safe queries
cursor.execute("CREATE TABLE accounts (id INTEGER, username TEXT)")
cursor.execute("INSERT INTO accounts VALUES (?, ?)", (1, "admin"))
# Simulate suspicious query (would be from user input in real scenario)
user_input = "admin' OR 1=1--"
try:
cursor.execute(f"SELECT * FROM accounts WHERE username = '{user_input}'")
except sqlite3.OperationalError:
pass
# Another suspicious pattern
cursor.execute("SELECT * FROM accounts WHERE username = 'admin' UNION SELECT 1, 2")
InjectionDetector 类针对已知的可疑模式检查每个 SQL 语句，这些模式可能表明 SQL 注入尝试。 找到时，它会打印警告。
虽然这是一个简单的示例，但它演示了如何将跟踪回调用于安全监视。 在生产中，您需要更复杂的检测和适当的警报。
将跟踪回调与连接钩子结合使用
此高级示例展示了如何将跟踪回调与其他 SQLite 连接钩子结合使用，以实现全面的监视。
combined_hooks.py
import sqlite3
class DatabaseMonitor:
def __init__(self):
self.total_queries = 0
def trace_callback(self, sql):
self.total_queries += 1
print(f"Query #{self.total_queries}: {sql}")
def progress_handler(self):
print("Database operation in progress...")
return 0 # Return non-zero to abort operation
def commit_hook(self):
print("Commit about to be executed")
return 0 # Return non-zero to abort commit
def rollback_hook(self):
print("Rollback occurred")
monitor = DatabaseMonitor()
with sqlite3.connect(":memory:") as conn:
# Set up all hooks
conn.set_trace_callback(monitor.trace_callback)
conn.set_progress_handler(monitor.progress_handler, 100)
conn.commit_hook = monitor.commit_hook
conn.rollback_hook = monitor.rollback_hook
cursor = conn.cursor()
# Perform operations that will trigger hooks
cursor.execute("CREATE TABLE test (id INTEGER)")
for i in range(150):
cursor.execute("INSERT INTO test VALUES (?)", (i,))
conn.commit()
# Force a rollback
try:
cursor.execute("INVALID SQL")
conn.commit()
except sqlite3.OperationalError:
conn.rollback()
print(f"Total queries executed: {monitor.total_queries}")
此示例将跟踪回调与 SQLite 的进度处理程序、提交钩子和回滚钩子结合在一起，以创建一个全面的监视解决方案。
进度处理程序在长时间操作期间定期调用，而提交和回滚钩子提供事务生命周期监视。 结合跟踪回调，这可以完全了解数据库活动。
最佳实践
- 保持回调快速： 它们在查询期间同步执行
- 用于调试： 考虑在生产环境中禁用
- 与日志记录结合使用： 用于持久跟踪记录
- 处理异常： 回调中的错误可能会导致您的应用崩溃
- 完成后重置： 在不需要时将回调设置为 None
资料来源
作者
列出所有 Python 教程。