我创建了一个sqlite3 db并成功地与它进行了事务处理。
一切正常。
我想深入了解python,所以我一直在搜索和发现
装饰工
和
上下文管理器
我试图在我的查询执行函数上实现这些概念。但是,我遇到了一个问题。
我创建了一个处理打开和关闭连接任务的类。
数据库上下文管理器.py
班级:
class DB_ContextManager():
def __init__(self, db_connection):
self.db_connection = db_connection
def __enter__(self):
self.conn = sqlite3.connect(self.db_connection)
return self.conn
def __exit__(self, exc_type, exc_val, exc_tb): # obligatory params
self.conn.close()
也创造了
连接b.py
它负责执行查询。
from Database.DB_ContextManager import DB_ContextManager as DB_CM
# Handles SELECT queries
def ExecuteSelectQuery(self, pQuery):
try:
with DB_CM(db_connection_string) as conn:
cur = conn.cursor()
cur.execute(pQuery)
result = cur.fetchall()
return result
except Exception as e:
LH.Handler(log_folder, 'ConnectionDB', 'Queries', 'ExecuteSelectQuery', e)
raise DE.ConnectionDB_Exception()
# Handles INSERTs, UPDATEs, DELETEs queries
def ExecuteNonQuery(self, pQuery):
try:
with DB_CM(db_connection_string) as conn:
cur = conn.cursor()
cur.execute(pQuery)
except Exception as e:
LH.Handler(log_folder, 'ConnectionDB', 'Queries', 'ExecuteSelectNonQuery', e)
raise DE.ConnectionDB_Exception()
如你所见
with DB_CM(db_connection_string) as conn:
cur = conn.cursor()
cur.execute(pQuery)
在每个函数中重复
为了避免这种情况,
我想创建一个decorator函数来封装这段代码
.
我的问题是,游标在ContextManager中“死亡”,例如,ExecuteSelectQuery需要游标在执行查询后获取返回数据。
我知道这是一个小项目,想这么长的时间在未来可能没有必要。但是,记住,这是一个开始的项目,我正在学习应用新的概念。
解决方案
正如@blhsing所建议的,我返回的是连接对象,而不是ContextManager中的光标。
我也处理
commit()
和
rollback()
在里面。
所以,总结一下:
连接b.py
def ExecuteSelectQuery(self, pQuery):
with DB_CM(db_connection_string, pQuery) as cur:
result = cur.fetchall()
return result
def ExecuteSelectNonQuery(self, pQuery):
with DB_CM(db_connection_string, pQuery) as cur:
pass
和
连接b.py
class DB_ContextManager():
def __init__(self, db_connection, pQuery):
self.db_connection = db_connection
self.query = pQuery
def __enter__(self):
try:
self.conn = sqlite3.connect(self.db_connection)
cur = self.conn.cursor()
cur.execute(self.query)
self.conn.commit()
return cur
except Exception as e:
LH.Handler(log_folder, 'DB_ContextManager', 'DB_ContextManager', '__enter__', e)
self.conn.rollback()
raise DE.ConnectionDB_Exception()
def __exit__(self, exc_type, exc_val, exc_tb): # obligatory params
self.conn.close()
