以可测试的方式连接到mongodb

Question

我打算使用Flask和MongoDB(可能是Ming作为ODM)在python中编写一个webapp.问题是我想保持我的模型和控制器真正分离,其中一个原因是能够在单独的组件上运行简单的单元测试.

现在这是我的问题,在我需要连接到MongoDB的请求生命周期中的某个时刻.每个请求都有一个单独的连接.Flask提供了一个线程本地对象,它可以包含请求的全局变量,这似乎是放置mongo连接的好地方.但是,这会在数据层和Flask之间产生一种硬依赖关系,这将使得单独测试或运行它们非常困难.

所以我的问题是,是否有一个优雅的解决方案.我自己想出了几个选项,但它们远非优雅.

首先,我可以给数据模块一个函数,告诉它从哪里获取连接对象.或者类似地将它交给一个可以用来获取新连接的函数.

第二个选项是创建一个类,模块可以使用它来获取与MongoDB的连接,然后创建此类的两个版本,一个使用Flask的全局对象,另一个只是明显连接到MongoDB.

这些对我来说似乎都不健壮或优雅,有没有办法更好地做到这一点？

Answer 1

一种方法可以是使用Python模块级单例模式.创建一个具有'conn'对象的模块,(仅使用普通的PyMongo)

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

然后只需为PyMongo集合创建一个包装器

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

这conn.end_request()将导致连接返回到池,并且在每个find_one()上它将从池中获得连接.别担心,它们是线程安全的.

现在你可以使用类似的模型了

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