使用上下文管理器连接到 sqlite3 数据库

Question

我想创建一个class用于数据库访问的数据库，其中我希望程序员可以选择在上下文管理器（语句with）中使用该类。所以我尝试使用以下代码：

class dbAccess:

    def __init__(self, fileName):
        self.conn = sql.connect(fileName)
        self.c = conn.cursor()

    def __enter__(self, fileName):
        self.conn = sql.connect(fileName)
        self.c = conn.cursor()

    def __exit__(self):
        self.conn.close()

现在我还希望用户能够在上下文管理器之外使用它。就像文件一样open()。所以我想知道如何知道是否self.conn已经连接到数据库，以便用户使用它时，他/她不会遇到冲突？基本上，我想做类似的事情：

    def __enter__(self, fileName):
        if not alreadyConnected():
            self.conn = sql.connect(fileName)
            self.c = conn.cursor()

那么我该如何找出这个alreadyConnected函数呢？

Answer 1

上述上下文处理程序片段的主要问题是该__enter__方法不会将实例返回到调用with块：

with dbAccess(fnam) as db:
    assert db, "sorry, db is None"  # this will raise

第二个connect()也是多余的。的工作上下文处理程序看起来像这样，也消除了对像（或者更好的是）sqlite3这样的方法的需要：alreadyConnectedalready_connected

from pathlib import Path
import sqlite3

class SQLite:
    """
    A minimal sqlite3 context handler that removes pretty much all
    boilerplate code from the application level.
    """

    def __init__(self, path: Path):
        self.path = path

    def __enter__(self):
        self.connection: sqlite3.Connection = sqlite3.connect(self.path)
        self.connection.row_factory = sqlite3.Row
        self.cursor: sqlite3.Cursor = self.connection.cursor()
        # do not forget this or you will not be able to use methods of the
        # context handler in your with block
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.connection.close()


if __name__ == "__main__":
    # just an example
    db_path = Path(".") / "local" / "twitter.sqlite"
    sql_statement = "select distinct author_id from tweets"

    # above context handler removes pretty much ALL boilerplate code
    # from application level code
    with SQLite(db_path) as db:
        db.cursor.execute(sql_statement)
        for row in db.cursor:
            print(row["author_id"])

    # sqlite3 also offers a context handler, BUT
    # a) you have to manage row_factory and cursor repetitively in
    #    the application code...
    with sqlite3.Connection(db_path) as db:
        db.row_factory = sqlite3.Row        # boilerplate
        cursor = db.cursor()                # boilerplate
        cursor.execute(sql_statement)
        for row in cursor:
            print(row["author_id"])
    # b) and it does NOT close connections on __exit__
    c = db.cursor()
    c.execute(sql_statement)
    for row in c:
        print(row["author_id"])

请注意，该sqlite3模块立即提供Connection上下文处理程序，但有一些缺点，如代码所示。