如何组织数据库访问层？

我正在使用SqlAlchemy，一个Python的ORM库。我曾经通过调用SqlAlchemy API直接从业务层直接访问数据库。如何组织数据库访问层？

但后来我发现，会造成太多的时间来运行我的所有测试用例，现在我想也许我应该创建一个数据库访问层，因此测试的，而不是直接访问数据库的过程中，我可以使用模拟对象。

我认为有两个选择这样做：

使用包含一个数据库连接和许多方法，如ADDUSER/delUser/UpdateUser两个，addBook/delBook/updateBook一个类。但这意味着这个类将会非常大。
另一种方法是创建一个像 “的UserManager”， “BookManager的” 不同的管理类。但这意味着我必须将经理列表传递给业务层，这似乎有点麻烦。

如何组织数据库层？

来源

2009-08-25 ablmf

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

组织代码，以便您可以测试大多数应用程序逻辑而无需访问数据库。这意味着每个类都有访问数据的方法和处理它的方法，而第二个类可以很容易地测试。
当您需要测试数据库访问时，您可以使用代理（如同解决方案＃1一样）;您可以将其视为SqlAlchemy的引擎，或者作为SA的直接替代品。在这两种情况下，您都可能想到self initializing fake。
如果代码不涉及存储过程，考虑使用内存数据库，像梅里说，（即使在这种情况下，把它称为“单元测试”听起来可能有点奇怪！）。

但是，从我的经验来看，一切都很简单，然后突然下降，当你在球场上。例如，当大多数逻辑在SQL语句中该怎么办？如果访问数据严格与其处理交错，该怎么办？有时候你可能会重构，有时候（特别是对于大型和遗留应用程序而言）不是。

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

来源

2009-08-25 08:05:45

我会在测试期间建立数据库连接，而不是连接到内存数据库。像这样：

sqlite_memory_db = create_engine('sqlite://')

这将是几乎一样快，你可以得到的，你也不能连接到一个真正的数据库，但只是暂时的一个在内存中，所以你不必担心测试后剩余的测试所做的更改等等，而且您不必嘲笑任何东西。

来源

2009-08-25 06:09:03

嗨，因为我的一些同事坚持我们应该使用存储过程，并且我没有控制权，所以sqlite对我来说不是一种可能的选择。 – ablmf 2009-08-25 07:33:16

顺便说一句：sqlite不支持存储过程。 – ablmf 2009-08-25 08:38:02

嗯。这意味着你将不得不模拟代码的重要部分（存储过程）。这将使测试更加有用。棘手的情况。 – 2009-08-25 10:55:04

SQLAlchemy的有making mocking easier一些设施 - 也许这将是比试图重写你的项目的整个部分更容易？

来源

2009-08-25 07:58:33 brool

感谢@brool，但现在这个链接已经被破坏了。 :( – 2015-05-14 19:47:18

一个捕捉修改数据库的方式，是用这样的使用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'

但是你要至少做一些测试与数据库反正因为你ATLEAST想知道如果你的查询正常工作。并且请记住，您还可以通过session.update()，.delete()和.execute()修改数据库。

来源

2009-08-25 08:51:23

如何组织数据库访问层？

回答

相关问题