Подключение к mongodb тестируемым способом - PullRequest
Новая
       27

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

4 голосов
/ 17 декабря 2011

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

Теперь вот моя проблема, в какой-то моментжизненный цикл запроса мне нужно для подключения к MongoDB.Каждый запрос будет иметь отдельное соединение.Flask предлагает локальный объект потока, который может содержать любые переменные, глобальные для запроса, это, кажется, хорошее место для установки соединения Монго.Однако это создает жесткую зависимость между уровнем данных и Flask, что усложнит тестирование или запуск их отдельно.

Поэтому мой вопрос на самом деле заключается в том, существует ли элегантное решение для этого.Я сам придумал пару вариантов, но они очень далеки от элегантности.

Сначала я мог бы просто дать модулю данных функцию, которая сообщит ему, откуда взять объект соединения.Или аналогичным образом передайте ему функцию, которую он может использовать для извлечения нового соединения.

Второй вариант - создать класс, который модуль может использовать для получения соединения с MongoDB, а затем создать 2 версии этого класса.тот, который использует глобальный объект Flask, а другой просто подключается к MongoDB.

Оба из них не кажутся мне действительно надежными или элегантными, есть ли способ сделать это лучше?

1 Ответ

5 голосов
/ 17 декабря 2011

Одним из подходов может быть использование одноэлементного шаблона уровня модуля 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'})
...