如何组织数据库访问层?

15 投票
4 回答
6606 浏览
提问于 2025-04-15 13:51

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

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

我觉得有两种选择可以实现这个目标:

  1. 第一种是使用一个单一的类,这个类包含一个数据库连接和很多方法,比如添加用户、删除用户、更新用户,添加书籍、删除书籍、更新书籍。但这样的话,这个会变得非常庞大。

  2. 第二种方法是创建不同的管理类,比如“用户管理器”、“书籍管理器”。但这样的话,我就得把一系列管理器传递给业务层,这样看起来有点麻烦。

你会怎么组织数据库层呢?

4 个回答

2

捕捉数据库的修改有一种方法,就是使用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()来修改数据库。

2

我会在测试的时候设置一个数据库连接,连接到一个内存数据库,而不是连接到真实的数据库。像这样:

sqlite_memory_db = create_engine('sqlite://')

这样做的速度会非常快,因为你并不是连接到一个真实的数据库,而是一个临时的内存数据库,所以你不用担心测试过程中做的修改会在测试后留下痕迹等等。而且你也不需要去模拟任何东西。

6

这是个好问题!
这个问题并不简单,可能需要用几种方法来解决。比如:

  1. 先整理代码,这样你就可以在不访问数据库的情况下测试大部分应用逻辑。这意味着每个类会有一些方法用来获取数据,还有一些方法用来处理数据,而后者就比较容易测试。
  2. 当你需要测试数据库访问时,可以使用一个代理(就像第一种解决方案那样);你可以把它想象成SqlAlchemy的引擎,或者是SA的替代品。在这两种情况下,你可能会想要考虑一个自初始化的假对象
  3. 如果代码不涉及存储过程,可以考虑使用内存数据库,正如Lennart所说的(即使在这种情况下,把它称为“单元测试”可能听起来有点奇怪!)。

不过,根据我的经验,纸上谈兵很简单,但一到实际操作就会遇到很多问题。比如,当大部分逻辑都在SQL语句里时该怎么办?如果数据访问和处理是紧密交织在一起的呢?有时候你可以重构代码,有时候(尤其是对于大型和遗留应用)就不行。

最后,我觉得这主要是一个心态的问题。
如果你认为需要有单元测试,并且希望它们运行得快,那么你就会以某种方式设计你的应用,以便更容易进行单元测试。
不幸的是,这并不总是成立(很多人认为单元测试可以在夜间运行,所以时间不是问题),结果你得到的东西可能根本无法进行单元测试。

撰写回答