时间戳

首页 > 解决方案 > 如何使用自己的功能扩展 sqlite3 连接对象？

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

问题描述

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

目前我有一个带有 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"

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

看来我应该写一个类扩展基本的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：如何成功继承 Sqlite3.Cursor 并添加我的自定义方法，但我似乎无法理解如何为我的目的使用类似的构造。

标签： pythonpython-2.7sqlite

解决方案

你不是那么远。

首先，类初始化器只能返回None（强调我的）：

因为__new__()和__init__()一起构造对象（__new__()创建对象和__init__()自定义对象），所以;不能返回非None值。__init__()这样做会导致 aTypeError在运行时引发。

其次，您使用初始化程序中self的Db对象覆盖对象的当前实例。sqlite3.Connection这使得 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 在问题的评论中说，诸如游标方法execute()之类的方法，根据DB-API 2.0规范，这是正确的。 但是，的连接对象提供了一些游标方法的快捷方式：
sqlite3

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

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

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

class Db(sqlite3.Connection):

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

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

当涉及到类时，该协议需要实现魔术方法__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")

有关上下文管理器的进一步阅读，请参阅官方文档。我还会推荐Jeff Knupp 的精彩介绍。此外，前面提到的PEP 343值得一看，以了解该协议背后的技术规范和基本原理。

推荐阅读