Tests

Привет,

Я создатель Dependency Injector. Это dependency injection фреймворк для Python.

Это еще одно руководство по построению приложений с помощью Dependency Injector.

Сегодня хочу показать как можно построить асинхронный демон на базе модуля asyncio.

Руководство состоит из таких частей:

1. Что мы будем строить?
2. Проверка инструментов
3. Структура проекта
4. Подготовка окружения
5. Логирование и конфигурация
6. Диспетчер
7. Мониторинг example.com
8. Мониторинг httpbin.org
9. Тесты
10. Заключение

Завершенный проект можно найти на Github.

Для старта желательно иметь:

• Начальные знания по asyncio
• Общее представление о принципе dependency injection

Что мы будем строить?

Мы будем строить мониторинг демон, который будет следить за доступом к веб-сервисам.

Демон будет посылать запросы к example.com и httpbin.org каждые несколько секунд. При получении ответа он будет записывать в лог такие данные:

• Код ответа
• Количество байт в ответе
• Время, затраченное на выполнение запроса

Проверка инструментов

Мы будем использовать Docker и docker-compose. Давайте проверим, что они установлены:

docker --versiondocker-compose --version

Вывод должен выглядеть приблизительно так:

Docker version 19.03.12, build 48a66213fedocker-compose version 1.26.2, build eefe0d31

Если Docker или docker-compose не установлены, их нужно установить перед тем как продолжить. Следуйте этим руководствам:

• Установка Docker
• Установка docker-compose

Инструменты готовы. Переходим к структуре проекта.

Структура проекта

Создаем папку проекта и переходим в нее:

mkdir monitoring-daemon-tutorialcd monitoring-daemon-tutorial

Теперь нам нужно создать начальную структуру проекта. Создаем файлы и папки следуя структуре ниже. Все файлы пока будут пустыми. Мы наполним их позже.

Начальная структура проекта:

./ monitoringdaemon/    __init__.py    __main__.py    containers.py config.yml docker-compose.yml Dockerfile requirements.txt

Начальная структура проекта готова. Мы расширим ее с следующих секциях.

Дальше нас ждет подготовка окружения.

Подготовка окружения

В этом разделе мы подготовим окружение для запуска нашего демона.

Для начала нужно определить зависимости. Мы будем использовать такие пакеты:

• dependency-injector dependency injection фреймворк
• aiohttp веб фреймворк (нам нужен только http клиент)
• pyyaml библиотека для парсинга YAML файлов, используется для чтения конфига
• pytest фреймворк для тестирования
• pytest-asyncio библиотека-помогатор для тестирования asyncio приложений
• pytest-cov библиотека-помогатор для измерения покрытия кода тестами

Добавим следующие строки в файл requirements.txt:

dependency-injectoraiohttppyyamlpytestpytest-asynciopytest-cov

И выполним в терминале:

pip install -r requirements.txt

Далее создаем Dockerfile. Он будет описывать процесс сборки и запуска нашего демона. Мы будем использовать python:3.8-buster в качестве базового образа.

Добавим следующие строки в файл Dockerfile:

FROM python:3.8-busterENV PYTHONUNBUFFERED=1WORKDIR /codeCOPY . /code/RUN apt-get install openssl \ && pip install --upgrade pip \ && pip install -r requirements.txt \ && rm -rf ~/.cacheCMD ["python", "-m", "monitoringdaemon"]

Последним шагом определим настройки docker-compose.

Добавим следующие строки в файл docker-compose.yml:

version: "3.7"services:  monitor:    build: ./    image: monitoring-daemon    volumes:      - "./:/code"

Все готово. Давайте запустим сборку образа и проверим что окружение настроено верно.

Выполним в терминале:

docker-compose build

Процесс сборки может занять несколько минут. В конце вы должны увидеть:

Successfully built 5b4ee5e76e35Successfully tagged monitoring-daemon:latest

После того как процесс сборки завершен запустим контейнер:

docker-compose up

Вы увидите:

Creating network "monitoring-daemon-tutorial_default" with the default driverCreating monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitoring-daemon-tutorial_monitor_1 exited with code 0

Окружение готово. Контейнер запускается и завершает работу с кодом 0.

Следующим шагом мы настроим логирование и чтение файла конфигурации.

Логирование и конфигурация

В этом разделе мы настроим логирование и чтение файла конфигурации.

Начнем с добавления основной части нашего приложения контейнера зависимостей (дальше просто контейнера). Контейнер будет содержать все компоненты приложения.

Добавим первые два компонента. Это объект конфигурации и функция настройки логирования.

Отредактируем containers.py:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )

Мы использовали параметры конфигурации перед тем как задали их значения. Это принцип, по которому работает провайдер Configuration.

Сначала используем, потом задаем значения.

Настройки логирования будут содержаться в конфигурационном файле.

Отредактируем config.yml:

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"

Теперь определим функцию, которая будет запускать наш демон. Её обычно называют main(). Она будет создавать контейнер. Контейнер будет использован для чтения конфигурационного файла и вызова функции настройки логирования.

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main() -> None:    """Run the application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.configure_logging()if __name__ == '__main__':    main()

Контейнер первый объект в приложении. Он используется для получения всех остальных объектов.

Логирование и чтение конфигурации настроено. В следующем разделе мы создадим диспетчер мониторинговых задач.

Диспетчер

Пришло время добавить диспетчер мониторинговых задач.

Диспетчер будет содержать список мониторинговых задач и контролировать их выполнение. Он будет выполнять каждую задачу в соответствии с расписанием. Класс Monitor базовый класс для мониторинговых задач. Для создания конкретных задач нужно добавлять дочерние классы и реализовывать метод check().


Добавим диспетчер и базовый класс мониторинговой задачи.

Создадим dispatcher.py и monitors.py в пакете monitoringdaemon:

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    monitors.py config.yml docker-compose.yml Dockerfile requirements.txt

Добавим следующие строки в файл monitors.py:

"""Monitors module."""import loggingclass Monitor:    def __init__(self, check_every: int) -> None:        self.check_every = check_every        self.logger = logging.getLogger(self.__class__.__name__)    async def check(self) -> None:        raise NotImplementedError()

и в файл dispatcher.py:

""""Dispatcher module."""import asyncioimport loggingimport signalimport timefrom typing import Listfrom .monitors import Monitorclass Dispatcher:    def __init__(self, monitors: List[Monitor]) -> None:        self._monitors = monitors        self._monitor_tasks: List[asyncio.Task] = []        self._logger = logging.getLogger(self.__class__.__name__)        self._stopping = False    def run(self) -> None:        asyncio.run(self.start())    async def start(self) -> None:        self._logger.info('Starting up')        for monitor in self._monitors:            self._monitor_tasks.append(                asyncio.create_task(self._run_monitor(monitor)),            )        asyncio.get_event_loop().add_signal_handler(signal.SIGTERM, self.stop)        asyncio.get_event_loop().add_signal_handler(signal.SIGINT, self.stop)        await asyncio.gather(*self._monitor_tasks, return_exceptions=True)        self.stop()    def stop(self) -> None:        if self._stopping:            return        self._stopping = True        self._logger.info('Shutting down')        for task, monitor in zip(self._monitor_tasks, self._monitors):            task.cancel()        self._logger.info('Shutdown finished successfully')    @staticmethod    async def _run_monitor(monitor: Monitor) -> None:        def _until_next(last: float) -> float:            time_took = time.time() - last            return monitor.check_every - time_took        while True:            time_start = time.time()            try:                await monitor.check()            except asyncio.CancelledError:                break            except Exception:                monitor.logger.exception('Error executing monitor check')            await asyncio.sleep(_until_next(last=time_start))

Диспетчер нужно добавить в контейнер.

Отредактируем containers.py:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            # TODO: add monitors        ),    )

Каждый компонент добавляется в контейнер.

В завершении нам нужно обновить функцию main(). Мы получим диспетчер из контейнера и вызовем его метод run().

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main() -> None:    """Run the application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.configure_logging()    dispatcher = container.dispatcher()    dispatcher.run()if __name__ == '__main__':    main()

Теперь запустим демон и проверим его работу.

Выполним в терминале:

docker-compose up

Вывод должен выглядеть так:

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 16:12:35,772] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutting downmonitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutdown finished successfullymonitoring-daemon-tutorial_monitor_1 exited with code 0

Все работает верно. Диспетчер запускается и выключается так как мониторинговых задач нет.

К концу этого раздела каркас нашего демона готов. В следующем разделе мы добавим первую мониторинговую задачу.

Мониторинг example.com

В этом разделе мы добавим мониторинговую задачу, которая будет следить за доступом к http://example.com.

Мы начнем с расширения нашей модели классов новым типом мониторинговой задачи HttpMonitor.

HttpMonitor это дочерний класс Monitor. Мы реализуем метод check(). Он будет отправлять HTTP запрос и логировать полученный ответ. Детали выполнения HTTP запроса будут делегированы классу HttpClient.


Сперва добавим HttpClient.

Создадим файл http.py в пакете monitoringdaemon:

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    http.py    monitors.py config.yml docker-compose.yml Dockerfile requirements.txt

И добавим в него следующие строки:

"""Http client module."""from aiohttp import ClientSession, ClientTimeout, ClientResponseclass HttpClient:    async def request(self, method: str, url: str, timeout: int) -> ClientResponse:        async with ClientSession(timeout=ClientTimeout(timeout)) as session:            async with session.request(method, url) as response:                return response

Далее нужно добавить HttpClient в контейнер.

Отредактируем containers.py:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            # TODO: add monitors        ),    )

Теперь мы готовы добавить HttpMonitor. Добавим его в модуль monitors.

Отредактируем monitors.py:

"""Monitors module."""import loggingimport timefrom typing import Dict, Anyfrom .http import HttpClientclass Monitor:    def __init__(self, check_every: int) -> None:        self.check_every = check_every        self.logger = logging.getLogger(self.__class__.__name__)    async def check(self) -> None:        raise NotImplementedError()class HttpMonitor(Monitor):    def __init__(            self,            http_client: HttpClient,            options: Dict[str, Any],    ) -> None:        self._client = http_client        self._method = options.pop('method')        self._url = options.pop('url')        self._timeout = options.pop('timeout')        super().__init__(check_every=options.pop('check_every'))    @property    def full_name(self) -> str:        return '{0}.{1}(url="{2}")'.format(__name__, self.__class__.__name__, self._url)    async def check(self) -> None:        time_start = time.time()        response = await self._client.request(            method=self._method,            url=self._url,            timeout=self._timeout,        )        time_end = time.time()        time_took = time_end - time_start        self.logger.info(            'Response code: %s, content length: %s, request took: %s seconds',            response.status,            response.content_length,            round(time_took, 3)        )

У нас все готово для добавления проверки http://example.com. Нам нужно сделать два изменения в контейнере:

• Добавить фабрику example_monitor.
• Передать example_monitor в диспетчер.

Отредактируем containers.py:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,        ),    )

Провайдер example_monitor имеет зависимость от значений конфигурации. Давайте добавим эти значения:

Отредактируем config.yml:

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"monitors:  example:    method: "GET"    url: "http://example.com"    timeout: 5    check_every: 5

Все готово. Запускаем демон и проверяем работу.

Выполняем в терминале:

docker-compose up

И видим подобный вывод:

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 17:06:41,965] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 17:06:42,033] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.067 secondsmonitor_1  |monitor_1  | [2020-08-08 17:06:47,040] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.073 seconds

Наш демон может следить за наличием доступа к http://example.com.

Давайте добавим мониторинг https://httpbin.org.

Мониторинг httpbin.org

В этом разделе мы добавим мониторинговую задачу, которая будет следить за доступом к http://example.com.

Добавление мониторинговой задачи для https://httpbin.org будет сделать легче, так как все компоненты уже готовы. Нам просто нужно добавить новый провайдер в контейнер и обновить конфигурацию.

Отредактируем containers.py:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    httpbin_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.httpbin,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,            httpbin_monitor,        ),    )

Отредактируем config.yml:

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"monitors:  example:    method: "GET"    url: "http://example.com"    timeout: 5    check_every: 5  httpbin:    method: "GET"    url: "https://httpbin.org/get"    timeout: 5    check_every: 5

Запустим демон и проверим логи.

Выполним в терминале:

docker-compose up

И видим подобный вывод:

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 18:09:08,540] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 18:09:08,618] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.077 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:08,722] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET https://httpbin.org/getmonitor_1  |     response code: 200monitor_1  |     content length: 310monitor_1  |     request took: 0.18 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:13,619] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.066 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:13,681] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET https://httpbin.org/getmonitor_1  |     response code: 200monitor_1  |     content length: 310monitor_1  |     request took: 0.126 seconds

Функциональная часть завершена. Демон следит за наличием доступа к http://example.com и https://httpbin.org.

В следующем разделе мы добавим несколько тестов.

Тесты

Было бы неплохо добавить несколько тестов. Давайте сделаем это.

Создаем файл tests.py в пакете monitoringdaemon:

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    http.py    monitors.py    tests.py config.yml docker-compose.yml Dockerfile requirements.txt

и добавляем в него следующие строки:

"""Tests module."""import asyncioimport dataclassesfrom unittest import mockimport pytestfrom .containers import ApplicationContainer@dataclasses.dataclassclass RequestStub:    status: int    content_length: int@pytest.fixturedef container():    container = ApplicationContainer()    container.config.from_dict({        'log': {            'level': 'INFO',            'formant': '[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s',        },        'monitors': {            'example': {                'method': 'GET',                'url': 'http://fake-example.com',                'timeout': 1,                'check_every': 1,            },            'httpbin': {                'method': 'GET',                'url': 'https://fake-httpbin.org/get',                'timeout': 1,                'check_every': 1,            },        },    })    return container@pytest.mark.asyncioasync def test_example_monitor(container, caplog):    caplog.set_level('INFO')    http_client_mock = mock.AsyncMock()    http_client_mock.request.return_value = RequestStub(        status=200,        content_length=635,    )    with container.http_client.override(http_client_mock):        example_monitor = container.example_monitor()        await example_monitor.check()    assert 'http://fake-example.com' in caplog.text    assert 'response code: 200' in caplog.text    assert 'content length: 635' in caplog.text@pytest.mark.asyncioasync def test_dispatcher(container, caplog, event_loop):    caplog.set_level('INFO')    example_monitor_mock = mock.AsyncMock()    httpbin_monitor_mock = mock.AsyncMock()    with container.example_monitor.override(example_monitor_mock), \            container.httpbin_monitor.override(httpbin_monitor_mock):        dispatcher = container.dispatcher()        event_loop.create_task(dispatcher.start())        await asyncio.sleep(0.1)        dispatcher.stop()    assert example_monitor_mock.check.called    assert httpbin_monitor_mock.check.called

Для запуска тестов выполним в терминале:

docker-compose run --rm monitor py.test monitoringdaemon/tests.py --cov=monitoringdaemon

Должен получиться подобный результат:

platform linux -- Python 3.8.3, pytest-6.0.1, py-1.9.0, pluggy-0.13.1rootdir: /codeplugins: asyncio-0.14.0, cov-2.10.0collected 2 itemsmonitoringdaemon/tests.py ..                                    [100%]----------- coverage: platform linux, python 3.8.3-final-0 -----------Name                             Stmts   Miss  Cover----------------------------------------------------monitoringdaemon/__init__.py         0      0   100%monitoringdaemon/__main__.py         9      9     0%monitoringdaemon/containers.py      11      0   100%monitoringdaemon/dispatcher.py      43      5    88%monitoringdaemon/http.py             6      3    50%monitoringdaemon/monitors.py        23      1    96%monitoringdaemon/tests.py           37      0   100%----------------------------------------------------TOTAL                              129     18    86%

Обратите внимание как в тесте test_example_monitor мы подменяем HttpClient моком с помощью метода .override(). Таким образом можно переопределить возвращаемое значения любого провайдера.

Такие же действия выполняются в тесте test_dispatcher для подмены моками мониторинговых задач.

Заключение

Мы построили мониторинг демон на базе asyncio применяя принцип dependency injection. Мы использовали Dependency Injector в качестве dependency injection фреймворка.

Преимущество, которое вы получаете с Dependency Injector это контейнер.

Контейнер начинает окупаться, когда вам нужно понять или изменить структуру приложения. С контейнером это легко, потому что все компоненты приложения и их зависимости в одном месте:

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    httpbin_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.httpbin,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,            httpbin_monitor,        ),    )

Контейнер как карта вашего приложения. Вы всегда знайте что от чего зависит.

Что дальше?

• Узнайте больше о Dependency Injector на GitHub
• Ознакомтесь с документацией на Read the Docs
• Есть вопрос или нашли баг? Откройте issue на Github

Привет,

Я создатель Dependency Injector. Это dependency injection фреймворк для Python.

Это завершающее руководство по построению приложений с помощью Dependency Injector. Прошлые руководства рассказывают как построить веб-приложение на Flask, REST API на Aiohttp и мониторинг демона на Asyncio применяя принцип dependency injection.

Сегодня хочу показать как можно построить консольное (CLI) приложение.

Дополнительно я подготовил ответы на часто задаваемые вопросы и опубликую их постскриптум.

Руководство состоит из таких частей:

1. Что мы будем строить?
2. Подготовка окружения
3. Структура проекта
4. Установка зависимостей
5. Фикстуры
6. Контейнер
7. Работа с csv
8. Работа с sqlite
9. Провайдер Selector
10. Тесты
11. Заключение
12. PS: вопросы и ответы

Завершенный проект можно найти на Github.

Для старта необходимо иметь:

• Python 3.5+
• Virtual environment

И желательно иметь общее представление о принципе dependency injection.

Что мы будем строить?

Мы будем строить CLI (консольное) приложение, которое ищет фильмы. Назовем его Movie Lister.

Как работает Movie Lister?

• У нас есть база данных фильмов
• О каждом фильме известна такая информация:
  
  • Название
  • Год выпуска
  • Имя режиссёра
• База данных распространяется в двух форматах:
  
  • Csv файл
  • Sqlite база данных
• Приложение выполняет поиск по базе данных по таким критериям:
  
  • Имя режиссёра
  • Год выпуска
• Другие форматы баз данных могут быть добавлены в будущем

Movie Lister это приложение-пример, которое используется в статье Мартина Фаулера о dependency injection и inversion of control.

Вот как выглядит диаграмма классов приложения Movie Lister:


Обязанности между классами распределены так:

• MovieLister отвечает за поиск
• MovieFinder отвечает за извлечение данных из базы
• Movie класс сущности фильм

Подготовка окружения

Начнём с подготовки окружения.

В первую очередь нам нужно создать папку проекта и virtual environment:

mkdir movie-lister-tutorialcd movie-lister-tutorialpython3 -m venv venv

Теперь давайте активируем virtual environment:

. venv/bin/activate

Окружение готово. Теперь займемся структурой проекта.

Структура проекта

В этом разделе организуем структуру проекта.

Создадим в текущей папке следующую структуру. Все файлы пока оставляем пустыми.

Начальная структура:

./ movies/    __init__.py    __main__.py    containers.py venv/ config.yml requirements.txt

Установка зависимостей

Пришло время установить зависимости. Мы будем использовать такие пакеты:

• dependency-injector dependency injection фреймворк
• pyyaml библиотека для парсинга YAML файлов, используется для чтения конфига
• pytest фреймворк для тестирования
• pytest-cov библиотека-помогатор для измерения покрытия кода тестами

Добавим следующие строки в файл requirements.txt:

dependency-injectorpyyamlpytestpytest-cov

И выполним в терминале:

pip install -r requirements.txt

Установка зависимостей завершена. Переходим к фикстурам.

Фикстуры

В это разделе мы добавим фикстуры. Фикстурами называют тестовые данные.

Мы создадим скрипт, который создаст тестовые базы данных.

Добавляем директорию data/ в корень проекта и внутрь добавляем файл fixtures.py:

./ data/    fixtures.py movies/    __init__.py    __main__.py    containers.py venv/ config.yml requirements.txt

Далее редактируем fixtures.py:

"""Fixtures module."""import csvimport sqlite3import pathlibSAMPLE_DATA = [    ('The Hunger Games: Mockingjay - Part 2', 2015, 'Francis Lawrence'),    ('Rogue One: A Star Wars Story', 2016, 'Gareth Edwards'),    ('The Jungle Book', 2016, 'Jon Favreau'),]FILE = pathlib.Path(__file__)DIR = FILE.parentCSV_FILE = DIR / 'movies.csv'SQLITE_FILE = DIR / 'movies.db'def create_csv(movies_data, path):    with open(path, 'w') as opened_file:        writer = csv.writer(opened_file)        for row in movies_data:            writer.writerow(row)def create_sqlite(movies_data, path):    with sqlite3.connect(path) as db:        db.execute(            'CREATE TABLE IF NOT EXISTS movies '            '(title text, year int, director text)'        )        db.execute('DELETE FROM movies')        db.executemany('INSERT INTO movies VALUES (?,?,?)', movies_data)def main():    create_csv(SAMPLE_DATA, CSV_FILE)    create_sqlite(SAMPLE_DATA, SQLITE_FILE)    print('OK')if __name__ == '__main__':    main()

Теперь выполним в терминале:

python data/fixtures.py

Скрипт должен вывести OK при успешном завершении.

Проверим, что файлы movies.csv и movies.db появились в директории data/:

./ data/    fixtures.py    movies.csv    movies.db movies/    __init__.py    __main__.py    containers.py venv/ config.yml requirements.txt

Фикстуры созданы. Продолжаем.

Контейнер

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

Контейнер позволяет описать структуру приложения в декларативном стиле. Он будет содержать все компоненты приложения их зависимости. Все зависимости будут указаны явно. Для добавления компонентов приложения в контейнер используются провайдеры. Провайдеры управляют временем жизни компонентов. При создании провайдера не происходит создание компонента. Мы указываем провайдеру как создавать объект, и он создаст его как только в этом будет необходимость. Если зависимостью одного провайдера является другой провайдер, то он будет вызван и так далее по цепочке зависимостей.

Отредактируем containers.py:

"""Containers module."""from dependency_injector import containersclass ApplicationContainer(containers.DeclarativeContainer):    ...

Контейнер пока пуст. Мы добавим провайдеры в следующих секциях.

Давайте еще добавим функцию main(). Её обязанность запускать приложение. Пока она будет только создавать контейнер.

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main():    container = ApplicationContainer()if __name__ == '__main__':    main()

Контейнер первый объект в приложении. Он используется для получения всех остальных объектов.

Работа с csv

Теперь добавим все что нужно для работы с csv файлами.

Нам понадобится:

• Сущность Movie
• Базовый класс MovieFinder
• Его реализация CsvMovieFinder
• Класс MovieLister

После добавления каждого компонента будем добавлять его в контейнер.



Создаем файл entities.py в пакете movies:

./ data/    fixtures.py    movies.csv    movies.db movies/    __init__.py    __main__.py    containers.py    entities.py venv/ config.yml requirements.txt

и добавляем внутрь следующие строки:

"""Movie entities module."""class Movie:    def __init__(self, title: str, year: int, director: str):        self.title = str(title)        self.year = int(year)        self.director = str(director)    def __repr__(self):        return '{0}(title={1}, year={2}, director={3})'.format(            self.__class__.__name__,            repr(self.title),            repr(self.year),            repr(self.director),        )

Теперь нам нужно добавить фабрику Movie в контейнер. Для этого нам понадобиться модуль providers из dependency_injector.

Отредактируем containers.py:

"""Containers module."""from dependency_injector import containers, providersfrom . import entitiesclass ApplicationContainer(containers.DeclarativeContainer):    movie = providers.Factory(entities.Movie)

Не забудьте убрать эллипсис (...). В контейнере уже есть провайдеры и он больше не нужен.

Переходим к созданию finders.

Создаем файл finders.py в пакете movies:

./ data/    fixtures.py    movies.csv    movies.db movies/    __init__.py    __main__.py    containers.py    entities.py    finders.py venv/ config.yml requirements.txt

и добавляем внутрь следующие строки:

"""Movie finders module."""import csvfrom typing import Callable, Listfrom .entities import Movieclass MovieFinder:    def __init__(self, movie_factory: Callable[..., Movie]) -> None:        self._movie_factory = movie_factory    def find_all(self) -> List[Movie]:        raise NotImplementedError()class CsvMovieFinder(MovieFinder):    def __init__(            self,            movie_factory: Callable[..., Movie],            path: str,            delimiter: str,    ) -> None:        self._csv_file_path = path        self._delimiter = delimiter        super().__init__(movie_factory)    def find_all(self) -> List[Movie]:        with open(self._csv_file_path) as csv_file:            csv_reader = csv.reader(csv_file, delimiter=self._delimiter)            return [self._movie_factory(*row) for row in csv_reader]

Теперь добавим CsvMovieFinder в контейнер.

Отредактируем containers.py:

"""Containers module."""from dependency_injector import containers, providersfrom . import finders, entitiesclass ApplicationContainer(containers.DeclarativeContainer):    config = providers.Configuration()    movie = providers.Factory(entities.Movie)    csv_finder = providers.Singleton(        finders.CsvMovieFinder,        movie_factory=movie.provider,        path=config.finder.csv.path,        delimiter=config.finder.csv.delimiter,    )

У CsvMovieFinder есть зависимость от фабрики Movie. CsvMovieFinder нуждается в фабрике так как будет создавать объекты Movie по мере того как будет читать данные из файла. Для того чтобы передать фабрику мы используем атрибут .provider. Это называется делегирование провайдеров. Если мы укажем фабрику movie как зависимость, она будет вызвана когда csv_finder будет создавать CsvMovieFinder и в качестве инъекции будет передан объект Movie. Используя атрибут .provider в качестве инъекции будет передам сам провайдер.

У csv_finder еще есть зависимость от нескольких опций конфигурации. Мы добавили провайдер Сonfiguration чтобы передать эти зависимости.

Мы использовали параметры конфигурации перед тем как задали их значения. Это принцип, по которому работает провайдер Configuration.

Сначала используем, потом задаем значения.

Теперь давайте добавим значения конфигурации.

Отредактируем config.yml:

finder:  csv:    path: "data/movies.csv"    delimiter: ","

Значения установлены в конфигурационный файл. Обновим функцию main() чтобы указать его расположение.

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main():    container = ApplicationContainer()    container.config.from_yaml('config.yml')if __name__ == '__main__':    main()

Переходим к listers.

Создаем файл listers.py в пакете movies:

./ data/    fixtures.py    movies.csv    movies.db movies/    __init__.py    __main__.py    containers.py    entities.py    finders.py    listers.py venv/ config.yml requirements.txt

и добавляем внутрь следующие строки:

"""Movie listers module."""from .finders import MovieFinderclass MovieLister:    def __init__(self, movie_finder: MovieFinder):        self._movie_finder = movie_finder    def movies_directed_by(self, director):        return [            movie for movie in self._movie_finder.find_all()            if movie.director == director        ]    def movies_released_in(self, year):        return [            movie for movie in self._movie_finder.find_all()            if movie.year == year        ]

Обновляем containers.py:

"""Containers module."""from dependency_injector import containers, providersfrom . import finders, listers, entitiesclass ApplicationContainer(containers.DeclarativeContainer):    config = providers.Configuration()    movie = providers.Factory(entities.Movie)    csv_finder = providers.Singleton(        finders.CsvMovieFinder,        movie_factory=movie.provider,        path=config.finder.csv.path,        delimiter=config.finder.csv.delimiter,    )    lister = providers.Factory(        listers.MovieLister,        movie_finder=csv_finder,    )

Все компоненты созданы и добавлены в контейнер.

В завершение обновляем функцию main().

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main():    container = ApplicationContainer()    container.config.from_yaml('config.yml')    lister = container.lister()    print(        'Francis Lawrence movies:',        lister.movies_directed_by('Francis Lawrence'),    )    print(        '2016 movies:',        lister.movies_released_in(2016),    )if __name__ == '__main__':    main()

Все готово. Теперь запустим приложение.

Выполним в терминале:

python -m movies

Вы увидите:

Francis Lawrence movies: [Movie(title='The Hunger Games: Mockingjay - Part 2', year=2015, director='Francis Lawrence')]2016 movies: [Movie(title='Rogue One: A Star Wars Story', year=2016, director='Gareth Edwards'), Movie(title='The Jungle Book', year=2016, director='Jon Favreau')]

Наше приложение работает с базой данных фильмов в формате csv. Нам нужно еще добавить поддержку формата sqlite. Разберемся с этим в следующем разделе.

Работа с sqlite

В это разделе мы добавим другой тип MovieFinder SqliteMovieFinder.

Отредактируем finders.py:

"""Movie finders module."""import csvimport sqlite3from typing import Callable, Listfrom .entities import Movieclass MovieFinder:    def __init__(self, movie_factory: Callable[..., Movie]) -> None:        self._movie_factory = movie_factory    def find_all(self) -> List[Movie]:        raise NotImplementedError()class CsvMovieFinder(MovieFinder):    def __init__(            self,            movie_factory: Callable[..., Movie],            path: str,            delimiter: str,    ) -> None:        self._csv_file_path = path        self._delimiter = delimiter        super().__init__(movie_factory)    def find_all(self) -> List[Movie]:        with open(self._csv_file_path) as csv_file:            csv_reader = csv.reader(csv_file, delimiter=self._delimiter)            return [self._movie_factory(*row) for row in csv_reader]class SqliteMovieFinder(MovieFinder):    def __init__(            self,            movie_factory: Callable[..., Movie],            path: str,    ) -> None:        self._database = sqlite3.connect(path)        super().__init__(movie_factory)    def find_all(self) -> List[Movie]:        with self._database as db:            rows = db.execute('SELECT title, year, director FROM movies')            return [self._movie_factory(*row) for row in rows]

Добавляем провайдер sqlite_finder в контейнер и указываем его в качестве зависимости для провайдера lister.

Отредактируем containers.py:

"""Containers module."""from dependency_injector import containers, providersfrom . import finders, listers, entitiesclass ApplicationContainer(containers.DeclarativeContainer):    config = providers.Configuration()    movie = providers.Factory(entities.Movie)    csv_finder = providers.Singleton(        finders.CsvMovieFinder,        movie_factory=movie.provider,        path=config.finder.csv.path,        delimiter=config.finder.csv.delimiter,    )    sqlite_finder = providers.Singleton(        finders.SqliteMovieFinder,        movie_factory=movie.provider,        path=config.finder.sqlite.path,    )    lister = providers.Factory(        listers.MovieLister,        movie_finder=sqlite_finder,    )

У провайдера sqlite_finder есть зависимость от опций конфигурации, которые мы еще не определили. Обновим файл конфигурации:

Отредактируем config.yml:

finder:  csv:    path: "data/movies.csv"    delimiter: ","  sqlite:    path: "data/movies.db"

Готово. Давайте проверим.

Выполняем в терминале:

python -m movies

Вы увидите:

Francis Lawrence movies: [Movie(title='The Hunger Games: Mockingjay - Part 2', year=2015, director='Francis Lawrence')]2016 movies: [Movie(title='Rogue One: A Star Wars Story', year=2016, director='Gareth Edwards'), Movie(title='The Jungle Book', year=2016, director='Jon Favreau')]

Наше приложение поддерживает оба формата базы данных: csv и sqlite. Каждый раз когда нам нужно изменить формат нам приходится менять код в контейнере. Мы улучшим это в следующем разделе.

Провайдер Selector

В этом разделе мы сделаем наше приложение более гибким.

Больше не нужно будет делать изменения в коде для переключения между csv и sqlite форматами. Мы реализуем переключатель на базе переменной окружения MOVIE_FINDER_TYPE:

• Когда MOVIE_FINDER_TYPE=csv приложения использует формат csv.
• Когда MOVIE_FINDER_TYPE=sqlite приложения использует формат sqlite.

В этом нам поможет провайдер Selector. Он выбирает провайдер на основе опции конфигурации (документация).

Отредактрируем containers.py:

"""Containers module."""from dependency_injector import containers, providersfrom . import finders, listers, entitiesclass ApplicationContainer(containers.DeclarativeContainer):    config = providers.Configuration()    movie = providers.Factory(entities.Movie)    csv_finder = providers.Singleton(        finders.CsvMovieFinder,        movie_factory=movie.provider,        path=config.finder.csv.path,        delimiter=config.finder.csv.delimiter,    )    sqlite_finder = providers.Singleton(        finders.SqliteMovieFinder,        movie_factory=movie.provider,        path=config.finder.sqlite.path,    )    finder = providers.Selector(        config.finder.type,        csv=csv_finder,        sqlite=sqlite_finder,    )    lister = providers.Factory(        listers.MovieLister,        movie_finder=finder,    )

Мы создали провайдер finder и указали его в качестве зависимости для провайдера lister. Провайдер finder выбирает между провайдерами csv_finder и sqlite_finder во время выполнения. Выбор зависит от значения переключателя.

Переключателем является опция конфигурации config.finder.type. Когда ее значение csv используется провайдер из ключа csv. Аналогично для sqlite.

Теперь нам нужно считать значение config.finder.type из переменной окружения MOVIE_FINDER_TYPE.

Отредактируем __main__.py:

"""Main module."""from .containers import ApplicationContainerdef main():    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.config.finder.type.from_env('MOVIE_FINDER_TYPE')    lister = container.lister()    print(        'Francis Lawrence movies:',        lister.movies_directed_by('Francis Lawrence'),    )    print(        '2016 movies:',        lister.movies_released_in(2016),    )if __name__ == '__main__':    main()

Готово.

Выполним в терминале следующие команды:

MOVIE_FINDER_TYPE=csv python -m moviesMOVIE_FINDER_TYPE=sqlite python -m movies

Вывод при выполнении каждой команды будет выглядеть так:

Francis Lawrence movies: [Movie(title='The Hunger Games: Mockingjay - Part 2', year=2015, director='Francis Lawrence')]2016 movies: [Movie(title='Rogue One: A Star Wars Story', year=2016, director='Gareth Edwards'), Movie(title='The Jungle Book', year=2016, director='Jon Favreau')]

В этом разделе познакомились с провайдером Selector. C помощью этого провайдера можно делать приложение более гибким. Значение переключателя может быть задано из любого источника: конфигурационного файла, словаря, другого провайдера.

Подсказка:
Переопределение значения конфигурации из другого провайдера позволяет реализовать перегрузку конфигурации в приложении без перезапуска, на горячую.
Для этого нужно использовать делегирование провайдеров и метод .override().

В следующем разделе добавим несколько тестов.

Тесты

В завершение добавим несколько тестов.

Создаём файл tests.py в пакете movies:

./ data/    fixtures.py    movies.csv    movies.db movies/    __init__.py    __main__.py    containers.py    entities.py    finders.py    listers.py    tests.py venv/ config.yml requirements.txt

и добавляем в него следующие строки:

"""Tests module."""from unittest import mockimport pytestfrom .containers import ApplicationContainer@pytest.fixturedef container():    container = ApplicationContainer()    container.config.from_dict({        'finder': {            'type': 'csv',            'csv': {                'path': '/fake-movies.csv',                'delimiter': ',',            },            'sqlite': {                'path': '/fake-movies.db',            },        },    })    return containerdef test_movies_directed_by(container):    finder_mock = mock.Mock()    finder_mock.find_all.return_value = [        container.movie('The 33', 2015, 'Patricia Riggen'),        container.movie('The Jungle Book', 2016, 'Jon Favreau'),    ]    with container.finder.override(finder_mock):        lister = container.lister()        movies = lister.movies_directed_by('Jon Favreau')    assert len(movies) == 1    assert movies[0].title == 'The Jungle Book'def test_movies_released_in(container):    finder_mock = mock.Mock()    finder_mock.find_all.return_value = [        container.movie('The 33', 2015, 'Patricia Riggen'),        container.movie('The Jungle Book', 2016, 'Jon Favreau'),    ]    with container.finder.override(finder_mock):        lister = container.lister()        movies = lister.movies_released_in(2015)    assert len(movies) == 1    assert movies[0].title == 'The 33'

Теперь запустим тестирование и проверим покрытие:

pytest movies/tests.py --cov=movies

Вы увидите:

platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1plugins: cov-2.10.0collected 2 itemsmovies/tests.py ..                                              [100%]---------- coverage: platform darwin, python 3.8.3-final-0 -----------Name                   Stmts   Miss  Cover------------------------------------------movies/__init__.py         0      0   100%movies/__main__.py        10     10     0%movies/containers.py       9      0   100%movies/entities.py         7      1    86%movies/finders.py         26     13    50%movies/listers.py          8      0   100%movies/tests.py           24      0   100%------------------------------------------TOTAL                     84     24    71%

Мы использовали метод .override() провайдера finder. Провайдер переопределяется моком. При обращении к провайдеру finder теперь будет возвращен переопределяющий мок.

Работа закончена. Теперь давайте подведем итоги.

Заключение

Мы построили консольное (CLI) приложение применяя принцип dependency injection. Мы использовали Dependency Injector в качестве dependency injection фреймворка.

Преимущество, которое вы получаете с Dependency Injector это контейнер.

Контейнер начинает окупаться, когда вам нужно понять или изменить структуру приложения. С контейнером это легко, потому что все компоненты приложения и их зависимости определены явно и в одном месте:

"""Containers module."""from dependency_injector import containers, providersfrom . import finders, listers, entitiesclass ApplicationContainer(containers.DeclarativeContainer):    config = providers.Configuration()    movie = providers.Factory(entities.Movie)    csv_finder = providers.Singleton(        finders.CsvMovieFinder,        movie_factory=movie.provider,        path=config.finder.csv.path,        delimiter=config.finder.csv.delimiter,    )    sqlite_finder = providers.Singleton(        finders.SqliteMovieFinder,        movie_factory=movie.provider,        path=config.finder.sqlite.path,    )    finder = providers.Selector(        config.finder.type,        csv=csv_finder,        sqlite=sqlite_finder,    )    lister = providers.Factory(        listers.MovieLister,        movie_finder=finder,    )

Контейнер как карта вашего приложения. Вы всегда знайте что от чего зависит.

PS: вопросы и ответы

В комментариях к прошлому руководству были заданы классные вопросы: зачем это нужно?, зачем нужен фреймворк?, чем фреймворк помогает в реализации?.

Я подготовил ответы:

Что такое dependency injection?

• это принцип который уменьшает связывание (coupling) и увеличивает сцепление (cohesion)

Зачем мне применять dependency injection?

• твой код становится более гибким, понятным и лучше поддается тестированию
• у тебя меньше проблем когда тебе нужно понять как он работает или изменить его

Как мне начать применять dependency injection?

• ты начинаешь писать код следуя принципу dependency injection
• ты регистрируешь все компоненты и их зависимости в контейнере
• когда тебе нужен компонент, ты получаешь его из контейнера

Зачем мне для этого фреймворк?

• тебе нужен фреймворк для того чтобы не создавать свой. Код создания объектов будет дублироваться и его тяжело будет менять. Для того чтобы этого не было тебе нужен контейнер.
• фреймворк дает тебе контейнер и провайдеры
• провайдеры управляют временем жизни объектов. Тебе понадобятся фабрики, синглтоны и объекты конфигурации
• контейнер служит коллекцией провайдеров

Какаю цену я плачу?

• тебе нужно явно указывать зависимости в контейнере
• это дополнительная работа
• это начнет приносить дивиденды когда проект начнет расти
• или через 2 недели после его завершения (когда ты забудешь какие решения принимал и какова структура проекта)

Концепция Dependency Injector

В дополнение опишу концепцию Dependency Injector как фреймворка.

Dependency Injector основан на двух принципах:

• Явное лучше неявного (PEP20).
• Не делать никакой магии с вашим кодом.

Чем Dependency Injector отличается от другим фреймворков?

• Нет автоматического связывания. Фреймворк не делает автоматического связывания зависимостей. Не используется интроспекция, связывание по именам аргументов и / или типам. Потому что явное лучше неявного (PEP20).
• Не загрязняет код вашего приложения. Ваше приложение не знает о наличии Dependency Injector и не зависит от него. Никаких @inject декораторов, аннотаций, патчинга или других волшебных трюков.

Dependency Injector предлагает простой контракт:

• Вы показываете фреймворку как собирать объекты
• Фреймворк их собирает

Сила Dependency Injector в его простоте и прямолинейности. Это простой инструмент для реализации мощного принципа.

Что дальше?

Если вы заинтересовались, но сомневайтесь, моя рекомендация такая:

Попробуйте применить этот подход на протяжении 2-х месяцев. Он неинтуитивный. Нужно время чтобы привыкнуть и прочувствовать. Польза стает ощутимой, когда проект вырастает до 30+ компонентов в контейнере. Если не понравится много не потеряйте. Если понравится получите существенное преимущество.

• Узнайте больше о Dependency Injector на GitHub
• Ознакомтесь с документацией на Read the Docs

Буду рад фидбеку и отвечу на вопросы в комментариях.