How to organize a Data Base Access layer?

Question

I am using SqlAlchemy, a python ORM library. And I used to access database directly from business layer directly by calling SqlAlchemy API.

But then I found that would cause too much time to run all my test cases and now I think maybe I should create a DB access layer, so I can use mock objects during test instead of access database directly.

I think there are 2 choices to do that :

1. use a single class which contains a DB connection and many methods like addUser/delUser/updateUser, addBook/delBook/updateBook. But this means this class will be very large.

2. Another approach is create different manager classes like "UserManager", "BookManager". But that means I have to pass a list of managers to Business layer, which seems a little Cumbersome.

How will you organize a database layer?

Answer 1

I would set up a database connection during testing that connects to a in memory database instead. Like so:

sqlite_memory_db = create_engine('sqlite://')

That will be pretty much as fast as you can get, you are also not connecting to a real database, but just a temporary one in memory, so you don't have to worry about the changes done by your tests remaining after the test, etc. And you don't have to mock anything.

Answer 2

SQLAlchemy has some facilities for making mocking easier -- maybe that would be easier than trying to rewrite whole sections of your project?

Answer 3

That's a good question!
The problem is not trivial, and may require several approaches to tackle it. For instance:

1. Organize the code, so that you can test most of the application logic without accessing the database. This means that each class will have methods for accessing data, and methods for processing it, and the second ones may be tested easily.
2. When you need to test database access, you may use a proxy (so, like solution #1); you can think of it as an engine for SqlAlchemy or as a drop-in replacement for the SA. In both cases, you may want to think to a self initializing fake.
3. If the code does not involve stored procedures, think about using in-memory databases, like Lennart says (even if in this case, calling it "unit test" may sound a bit strange!).

However, from my experience, everything is quite easy on word, and then falls abruptly when you go on the field. For instance, what to do when most of the logic is in the SQL statements? What if accessing data is strictly interleaved with its processing? Sometimes you may be able to refactor, sometimes (especially with large and legacy applications) not.

In the end, I think it is mostly a matter of mindset.
If you think you need to have unit tests, and you need to have them running fast, then you design your application in a certain way, that allow for easier unit testing.
Unfortunately, this is not always true (many people see unit tests as something that can run overnight, so time is not an issue), and you get something that will not be really unit-testable.

Answer 4

One way to capture modifications to the database, is to use the SQLAlchemy session extension mechanism and intercept flushes to the database using something like this:

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 = {}

Then for tests you can configure your session with that mock and see if it matches your expectations.

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'

But you'll want to do at least some tests with a database anyway because you'll atleast want to know if your queries work as expected. And keep in mind that you can also have modifications to the database via session.update(), .delete() and .execute().