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

我创建了一个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()