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

Question

我是 Python 的新手，我正在尝试构建一个开始项目以进入该语言。

我创建了一个 SQLite3 数据库并设法与它进行交易。 一切正常。

我想更深入地了解 Python，所以我一直在搜索并发现 Decorators 和 Context Manager，并尝试在我的 Query Execution 函数中实现这些概念。但是，我遇到了一个问题。

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

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 所建议的，我在 ContextManager 中返回连接对象而不是光标。

我也在其中处理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()

这样ExecuteSelectQuery的try块可以修改为：

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

ExecuteNonQuery 的 try 块可以很简单：

with DB_CM(db_connection_string):
    pass