一、SQLite异常处理的重要性
在开发数据访问层时，异常处理是保证系统稳定性的关键。SQLite虽然轻量，但如果不妥善处理异常，可能会导致数据丢失、连接泄漏甚至应用崩溃。想象一下，用户正在提交重要数据，突然程序因为一个未捕获的SQL异常而退出，这体验得多糟糕？
SQLite的常见异常包括：
- 数据库文件被占用或损坏
- SQL语法错误
- 并发访问冲突
- 磁盘空间不足
我们来看一个反面教材（技术栈：Python + sqlite3）：
import sqlite3
# 错误示例：没有异常处理的危险代码
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO users (name) VALUES ('张三')") # 如果表不存在就崩溃
conn.commit()
conn.close()
二、基础异常处理框架
让我们构建一个安全的数据库操作模板（技术栈：Python）：
import sqlite3
from contextlib import contextmanager
@contextmanager
def get_db_connection():
"""上下文管理器自动处理连接生命周期"""
conn = None
try:
conn = sqlite3.connect('app.db', timeout=10)
conn.execute("PRAGMA foreign_keys = ON") # 启用外键约束
yield conn
except sqlite3.Error as e:
print(f"数据库连接失败: {e}")
raise
finally:
if conn:
conn.close()
def safe_execute(sql, params=None):
"""安全执行SQL语句的封装"""
try:
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(sql, params or ())
conn.commit()
return cursor.rowcount
except sqlite3.OperationalError as e:
print(f"操作失败: {e}")
return -1
except sqlite3.IntegrityError as e:
print(f"数据完整性错误: {e}")
return -2
这个模板解决了几个关键问题：
- 使用上下文管理器确保连接总是关闭
- 区分不同类型的SQLite错误
- 提供统一的错误代码返回值
三、高级异常处理策略
3.1 事务回滚机制
def transfer_funds(sender, receiver, amount):
"""带事务的资金转账示例"""
try:
with get_db_connection() as conn:
conn.execute("BEGIN TRANSACTION")
# 检查发送方余额
cursor = conn.execute("SELECT balance FROM accounts WHERE user_id=?", (sender,))
if cursor.fetchone()[0] < amount:
raise ValueError("余额不足")
# 执行转账
conn.execute("UPDATE accounts SET balance=balance-? WHERE user_id=?", (amount, sender))
conn.execute("UPDATE accounts SET balance=balance+? WHERE user_id=?", (amount, receiver))
conn.commit()
return True
except (sqlite3.Error, ValueError) as e:
conn.rollback() # 关键的回滚操作
print(f"转账失败: {e}")
return False
3.2 自定义异常类
class DBException(Exception):
"""自定义数据库异常基类"""
pass
class TableExistsError(DBException):
"""表已存在异常"""
pass
def create_table_safe():
"""安全建表示例"""
try:
with get_db_connection() as conn:
conn.execute("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT)")
except sqlite3.OperationalError as e:
if "already exists" in str(e):
raise TableExistsError("用户表已存在") from e
raise
四、实战：完整的DAO层实现
下面是一个完整的用户数据访问对象实现（技术栈：Python）：
class UserDAO:
"""用户数据访问对象"""
def __init__(self, db_file="app.db"):
self.db_file = db_file
def _execute(self, sql, params=None):
"""内部执行方法"""
try:
with sqlite3.connect(self.db_file) as conn:
cursor = conn.cursor()
cursor.execute(sql, params or ())
conn.commit()
return cursor
except sqlite3.Error as e:
print(f"数据库操作失败: {e}")
raise DBException("数据库操作失败") from e
def create_user(self, name, email):
"""创建用户"""
try:
self._execute(
"INSERT INTO users (name, email) VALUES (?, ?)",
(name, email)
)
return True
except sqlite3.IntegrityError:
raise ValueError("邮箱已存在")
def get_user(self, user_id):
"""获取单个用户"""
cursor = self._execute(
"SELECT id, name, email FROM users WHERE id=?",
(user_id,)
)
if row := cursor.fetchone():
return {"id": row[0], "name": row[1], "email": row[2]}
raise ValueError("用户不存在")
def search_users(self, keyword):
"""搜索用户"""
cursor = self._execute(
"SELECT id, name FROM users WHERE name LIKE ?",
(f"%{keyword}%",)
)
return [{"id": row[0], "name": row[1]} for row in cursor]
五、性能与可靠性平衡
5.1 连接池优化
import sqlite3
from queue import Queue
class SQLiteConnectionPool:
"""简单的SQLite连接池"""
def __init__(self, db_file, pool_size=5):
self.db_file = db_file
self.pool = Queue(pool_size)
for _ in range(pool_size):
conn = sqlite3.connect(db_file)
self.pool.put(conn)
def get_conn(self):
"""获取连接"""
return self.pool.get()
def release_conn(self, conn):
"""释放连接"""
self.pool.put(conn)
def close_all(self):
"""关闭所有连接"""
while not self.pool.empty():
conn = self.pool.get()
conn.close()
5.2 重试机制
import time
from functools import wraps
def retry_on_deadlock(max_retries=3, delay=0.1):
"""死锁重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except sqlite3.OperationalError as e:
if "database is locked" in str(e):
retries += 1
time.sleep(delay * retries)
continue
raise
raise Exception("超过最大重试次数")
return wrapper
return decorator
六、最佳实践总结
- 防御性编程：总是假设数据库操作可能失败
- 资源管理：使用with语句或RAII模式确保资源释放
- 错误分类：区分业务错误和技术错误
- 日志记录：记录足够多的上下文信息以便排查问题
- 单元测试：为异常场景编写专门的测试用例
# 测试用例示例
import unittest
class TestUserDAO(unittest.TestCase):
def setUp(self):
self.dao = UserDAO(":memory:")
self.dao._execute("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT, email TEXT UNIQUE)")
def test_create_duplicate_email(self):
self.dao.create_user("张三", "zhang@example.com")
with self.assertRaises(ValueError):
self.dao.create_user("李四", "zhang@example.com")
def test_get_nonexistent_user(self):
with self.assertRaises(ValueError):
self.dao.get_user(999)
评论