如何使用自己的功能扩展 sqlite3 连接对象？

Question

我有一个用 Python 2.7 编写的项目，其中主程序需要频繁访问 sqlite3 数据库以写入日志、测量结果、获取设置...

目前我有一个带有 add_log()、get_setting() 等函数的 db 模块，其中的每个函数基本上如下所示：

def add_log(logtext):
    try:
        db = sqlite3.connect(database_location)
    except sqlite3.DatabaseError as e:
        db.close()  # try to gracefully close the db
        return("ERROR (ADD_LOG): While opening db: {}".format(e))
    try:
        with db:  # using context manager to automatically commit or roll back changes.
            # when using the context manager, the execute function of the db should be used instead of the cursor
            db.execute("insert into logs(level, source, log) values (?, ?, ?)", (level, source, logtext))
    except sqlite3.DatabaseError as e:
        return("ERROR (ADD_LOG): While adding log to db: {}".format(e))
    return "OK"

（删除了一些额外的代码和 cmets）。

看来我应该写一个类扩展基本的sqlite连接对象功能，以便连接只创建一次（在主程序的开头），然后这个对象包含诸如

之类的功能

class Db(sqlite3.Connection):
    def __init__(self, db_location = database_location):
        try:
            self = sqlite3.connect(db_location)
            return self
        except sqlite3.DatabaseError as e:
            self.close()  # try to gracefully close the db

    def add_log(self, logtext):
        self.execute("insert into logs(level, source, log) values (?, ?, ?)", (level, source, logtext))

似乎这应该相当简单，但我似乎无法让它工作。

这里似乎有一些有用的建议： Python: How to successfully inherit Sqlite3.Cursor and add my customized method 但我似乎无法理解如何为我的目的使用类似的构造。

Answer 1

你并没有那么远。

首先，一个类初始化器cannot return anything but None（强调我的）：

因为__new__() 和__init__() 在构造对象时协同工作（__new__() 用于创建对象，__init__() 用于自定义对象），__init__() 不会返回非None 的值;这样做会导致在运行时引发TypeError。

其次，在初始化程序中使用 sqlite3.Connection 对象覆盖 Db 对象的当前实例 self。这使得继承 SQLite 的连接对象有点毫无意义。

您只需要修复您的 __init__ 方法即可使其正常工作：

class Db(sqlite3.Connection):

    # If you didn't use the default argument, you could omit overriding __init__ alltogether
    def __init__(self, database=database_location, **kwargs):
        super(Db, self).__init__(database=database, **kwargs)

    def add_log(self, logtext, level, source):
        self.execute("insert into logs(level, source, log) values (?, ?, ?)", (level, source, logtext))

这使您可以将类的实例用作上下文管理器：

with Db() as db:
    print [i for i in db.execute("SELECT * FROM logs")]
    db.add_log("I LAUNCHED THAT PUG INTO SPACE!", 42, "Right there")

---

Maurice Meyer 在问题的 cmets 中表示，execute() 等方法是 cursor methods，并且根据 DB-API 2.0 规范，这是正确的。
但是，sqlite3 的连接对象为游标方法提供了一些 shortcuts：

这是一个非标准的快捷方式，它通过调用游标方法创建一个中间游标对象，然后使用给定的参数调用游标的execute方法。

---

在 cmets 中展开讨论：
上面我的代码示例中关于默认参数的注释是针对覆盖sqlite3.Connection的__init__方法的要求。

Db 类中的 __init__ 只需要为 sqlite3.Connection 初始化程序的 database 参数定义默认值 database_location。
如果您愿意在该类的每个实例化时传递这样的值，那么您的自定义连接类可能看起来像这样，并且仍然以相同的方式工作，除了那个参数：

class Db(sqlite3.Connection):

    def add_log(self, logtext, level, source):
        self.execute("insert into logs(level, source, log) values (?, ?, ?)", (level, source, logtext))

但是，__init__ 方法与PEP 343 中定义的上下文管理器协议无关。

对于类，这个协议需要实现魔术方法__enter__和__exit__

sqlite3.Connection 遵循以下原则：

class Connection:

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_val is None:
            self.commit()
        else:
            self.rollback()

注意：sqlite3.Connection 由 C 模块提供，因此没有 Python 类定义。以上反映了这些方法的大致情况。

假设您不希望始终保持同一个连接打开，而是希望每个事务都有一个专用连接，同时保持上述Db 类的通用接口。
你可以这样做：

# Keep this to have your custom methods available
class Connection(sqlite3.Connection):

    def add_log(self, level, source, log):
        self.execute("INSERT INTO logs(level, source, log) VALUES (?, ?, ?)", 
                     (level, source, log))


class DBM:

    def __init__(self, database=database_location):
        self._database = database
        self._conn = None

    def __enter__(self):
        return self._connection()

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Decide whether to commit or roll back
        if exc_val:
            self._connection().rollback()
        else:
            self._connection().commit()
        # close connection
        try:
            self._conn.close()
        except AttributeError:
            pass
        finally:
            self._conn = None

    def _connection(self):
        if self._conn is None:
            # Instantiate your custom sqlite3.Connection
            self._conn = Connection(self._database)
        return self._conn

    # add shortcuts to connection methods as seen fit
    def execute(self, sql, parameters=()):
        with self as temp:
            result = temp.execute(sql, parameters).fetchall()
        return result

    def add_log(self, level, source, log):
        with self as temp:
            temp.add_log(level, source, log)

这可以在上下文中使用，也可以通过调用实例上的方法来使用：

db = DBM(database_location)

with db as temp:
    print [i for i in temp.execute("SELECT * FROM logs")]
    temp.add_log(1, "foo", "I MADE MASHED POTATOES")

# The methods execute and add_log are only available from
# the outside because the shortcuts have been added to DBM
print [i for i in db.execute("SELECT * FROM logs")]
db.add_log(1, "foo", "I MADE MASHED POTATOES")

---

有关上下文管理器的进一步阅读，请参阅official documentation。我也会推荐Jeff Knupp's nice introduction。此外，前面提到的PEP 343 值得一看，以了解该协议背后的技术规范和基本原理。