Подключение к mongodb тестируемым способом

Question

Я планирую написать веб-приложение на python, используя Flask и MongoDB (и, возможно, Ming как ODM). Проблема в том, что я хотел бы, чтобы моя модель и контроллер были действительно хорошо отделены друг от друга, одной из причин этого была бы возможность запускать простые юнит-тесты на отдельных компонентах.

Теперь вот моя проблема, в какой-то момент в жизненном цикле запроса мне нужно подключиться к MongoDB. Каждый запрос будет иметь отдельное соединение. Flask предлагает локальный объект потока, который может содержать любые переменные, которые являются глобальными для запроса, это, кажется, хорошее место, чтобы установить соединение Монго. Однако это создает жесткую зависимость между уровнем данных и 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'})