以可测试的方式连接到 mongodb答案

【问题标题】：Connecting to mongodb in a testable way以可测试的方式连接到 mongodb
【发布时间】：2012-01-22 13:03:51
【问题描述】：

我打算使用 Flask 和 MongoDB（可能还有 Ming 作为 ODM）在 python 中编写一个 webapp。问题是我想让我的模型和控制器很好地分开，这样做的一个原因是能够在单独的组件上运行简单的单元测试。

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

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

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

第二种选择是创建一个模块可以用来连接到 MongoDB 的类，然后创建这个类的 2 个版本，一个使用 Flask 的全局对象，另一个只是简单地连接到 MongoDB。

这两个对我来说似乎都不是很健壮或优雅，有什么方法可以做得更好吗？

【问题讨论】：

标签： python mongodb flask pymongo ming

【解决方案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'})

【讨论】：

但这并没有真正将模型与 Flask 解耦，因为它仍然需要 Flask 来进行 app.config，对吧？
app.config 用于获取 DB_NAME，如您所见。如果你愿意，你可以搭上它。