如何組織數據庫訪問層？

我正在使用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

如何組織數據庫訪問層？

回答

相關問題