python - 使用上下文管理器连接到 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 block :

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

第二个 connect() 也是多余的。 sqlite3 的工作上下文处理程序看起来像这样，同时也不再需要像 alreadyConnected(或者更好的是 already_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 作为上下文处理程序，但如代码中所示，存在一些缺点。