Python - 在上下文管理器外部管理游标连接

Question

我是Python的新手，我正在尝试构建一个启动项目来学习这种语言。

我创建了一个SQLite3 DB并设法用它进行交易。 一切正常。

我想更深入地了解Python，所以我一直在搜索并发现了装饰器和上下文管理器，我试图在我的查询执行函数中实现这些概念。但是，我遇到了一个问题。

我创建了一个处理打开和关闭连接任务的类。

DB_ContextManager.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()

并且还创建了负责执行查询的 ConnectionDB.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)

在每个函数中重复

为了避免这种情况，我想创建一个封装这段代码的装饰器函数。 我的问题是游标在 ContextManager 中"死亡"，例如，ExecuteSelectQuery 需要游标在执行查询后获取返回数据。

我知道这是一个小项目，未来可能没有必要考虑这么长远。但是，请记住，这是一个开始的项目，我正在学习应用新概念。

---

溶液

正如@blhsing建议的那样，我返回连接对象而不是上下文管理器中的光标。

我也在其中处理commit()和rollback()。

所以，总结一下：

ConnectionDB.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

和ConnectionDB.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()

Answer 1

您可以使上下文管理器返回游标而不是连接对象：

class DB_CM():
def __init__(self, db_connection):
self.db_connection = db_connection
def __enter__(self):
self.conn = sqlite3.connect(self.db_connection)
cur = self.conn.cursor()
cur.execute(pQuery)
return cur
def __exit__(self, exc_type, exc_val, exc_tb): # obligatory params
self.conn.close()

以便try块的ExecuteSelectQuery可以修改为：

with DB_CM(db_connection_string) as cur:
result = cur.fetchall()        
return result

ExecuteNonQuery的try块可以简单地：

with DB_CM(db_connection_string):
pass