如何组织数据库访问层？

Posted 2023-04-12

【中文标题】如何组织数据库访问层？

【英文标题】：How to organize a Data Base Access layer?

【发布时间】：2009-08-25 05:35:09

【问题描述】：

我正在使用 SqlAlchemy，一个 Python ORM 库。而我以前是通过调用SqlAlchemy API直接从业务层直接访问数据库的。

但后来我发现这会导致运行所有测试用例的时间过多，现在我想也许我应该创建一个数据库访问层，这样我就可以在测试期间使用模拟对象而不是直接访问数据库。

我认为有两种选择：

使用包含数据库连接和许多方法（如 addUser/delUser/updateUser、addBook/delBook/updateBook）的单个类。但这意味着这个类会非常大。

另一种方法是创建不同的管理器类，例如“UserManager”、“BookManager”。但这意味着我必须将管理器列表传递给业务层，这似乎有点麻烦。

您将如何组织数据库层？

【参考方案1】：

这是个好问题！ 这个问题并非微不足道，可能需要多种方法来解决。 例如：

组织代码，这样就可以在不访问数据库的情况下测试大部分应用程序逻辑。这意味着每个类都有访问数据的方法和处理数据的方法，并且可以轻松测试第二个类。 当您需要测试数据库访问时，您可以使用代理（因此，如解决方案 #1）；您可以将其视为 SqlAlchemy 的引擎或 SA 的替代品。在这两种情况下，您可能需要考虑self initializing fake。 如果代码不涉及存储过程，请考虑使用内存数据库，就像 Lennart 所说的那样（即使在这种情况下，将其称为“单元测试”可能听起来有点奇怪！）。

但是，根据我的经验，一切都很容易说出来，然后在你上场时突然跌倒。例如，当大部分逻辑都在 SQL 语句中时该怎么办？如果访问数据与其处理严格交错怎么办？有时您可能能够进行重构，有时（尤其是对于大型和遗留应用程序）则不能。

最后，我认为这主要是心态的问题。 如果您认为需要进行单元测试，并且需要让它们快速运行，那么您可以以某种方式设计您的应用程序，以便更轻松地进行单元测试。 不幸的是，这并不总是正确的（许多人认为单元测试可以在一夜之间运行，所以时间不是问题），而且你得到的东西并不是真正可单元测试的。

【参考方案2】：

我会在测试期间建立一个连接到内存数据库的数据库连接。像这样：

sqlite_memory_db = create_engine('sqlite://')

这将尽可能快，您也没有连接到真正的数据库，而只是内存中的一个临时数据库，因此您不必担心测试后剩余的更改测试等等。而且你不必嘲笑任何东西。

【讨论】：

你好，因为我的一些同事坚持我们应该使用存储过程，我无法控制，所以 sqlite 对我来说不是一个可能的选择。

顺便说一句：sqlite 不支持存储过程。

嗯。这意味着您将不得不模拟代码的重要部分（存储过程）。这将使测试变得不那么有用。棘手的情况。

【参考方案3】：

捕获对数据库的修改的一种方法是使用 SQLAlchemy 会话扩展机制并使用以下方式拦截对数据库的刷新：

from sqlalchemy.orm.attributes import instance_state
from sqlalchemy.orm import SessionExtension

class MockExtension(SessionExtension):
    def __init__(self):
        self.clear()

    def clear(self):
        self.updates = set()
        self.inserts = set()
        self.deletes = set()

    def before_flush(self, session, flush_context, instances):
        for obj in session.dirty:
            self.updates.add(obj)
            state = instance_state(obj)
            state.commit_all()
            session.identity_map._mutable_attrs.discard(state)
            session.identity_map._modified.discard(state)

        for obj in session.deleted:
            self.deletes.add(obj)
            session.expunge(obj)

        self.inserts.update(session.new)
        session._new = 

然后对于测试，您可以使用该模拟配置您的会话，看看它是否符合您的期望。

mock = MockExtension()
Session = sessionmaker(extension=[mock], expire_on_commit=False)

def do_something(attr):
    session = Session()
    obj = session.query(Cls).first()
    obj.attr = attr
    session.commit()

def test_something():
    mock.clear()
    do_something('foobar')
    assert len(mock.updates) == 1
    updated_obj = mock.updates.pop()
    assert updated_obj.attr == 'foobar'

但是无论如何，您至少需要对数据库进行一些测试，因为您至少想知道您的查询是否按预期工作。请记住，您还可以通过session.update()、.delete() 和.execute() 对数据库进行修改。

【参考方案4】：

SQLAlchemy 为making mocking easier 提供了一些工具——也许这比尝试重写项目的整个部分更容易？

【讨论】：

感谢@brool，但该链接现在已断开。 :(