Connecting to mongodb in a testable way

Question

I'm planning to write a webapp in python using Flask and MongoDB (and probably Ming as ODM). The problem is I would like to keep my model and controller really well separated, one reason for this would be to be able to run simple unittests on the separate components.

Now here is my problem, at some point in the request lifecycle I need to connect to MongoDB. Each request will have a separate connection. Flask offers a thread local object that can contain any variables that are global to the request, this seems to be a good place to put the mongo connection. However, this creates a hard dependency between the data layer and Flask, which will make testing or running them separately quite hard.

So my question really is whether or not there is an elegant solution for this. I've come up with a couple of options myself, but they are very far from elegant.

First I could just give the data module a function that would tell it where to get the connection object from. Or similarly hand it a function that it can use to fetch a new connection.

The second option is to create a class that the module can use to get a connection to MongoDB, and then create 2 versions of this class, one that uses Flask's global object, and the other just plainly connects to MongoDB.

Both of these don't seem really robust or elegant to me, is there any way to do this better?

Answer 1

One approach could be to make use of Python module level singleton pattern. Create a module having 'conn' object, (using just plain PyMongo)

try:
    conn = Connection(max_pool_size=20)
except ConnectionFailure:
    app.logger.critical("Unable to connect to MongoDB")

and then just create a wrapper for PyMongo collection

class Model(object):
    def __init__(self, table, db = app.config['DB_NAME']):
        self._table = table
        self._db = db

    def find_one(self, spec_or_id=None, *args, **kwargs):
        result = None
        try:
            result = conn[self._db][self._table].find_one(spec_or_id, *args, **kwargs)
        except InvalidName:
            app.logger.critical('invalid DB or Table name')
        finally:
            conn.end_request()

        return result

Here conn.end_request() will cause connection to return to the pool and on each find_one() it will get the connection from pool. Don't worry, they are thread safe.

now you can use the model something like

result = Model(COLLECTION).find_one({user:'joe'})