Aiohttp Dependency Injector руководство по применению dependency injection

Привет,

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

Продолжаю серию руководств по применению Dependency Injector для построения приложений.

В этом руководстве хочу показать как применять Dependency Injector для разработки aiohttp приложений.

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

1. Что мы будем строить?
2. Подготовка окружения
3. Структура проекта
4. Установка зависимостей
5. Минимальное приложение
6. Giphy API клиент
7. Сервис поиска
8. Подключаем поиск
9. Немного рефакторинга
10. Добавляем тесты
11. Заключение

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

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

• Python 3.5+
• Virtual environment

И желательно иметь:

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

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

Мы будем строить REST API приложение, которое ищет забавные гифки на Giphy. Назовем его Giphy Navigator.

Как работает Giphy Navigator?

• Клиент отправляет запрос указывая что искать и сколько результатов вернуть.
• Giphy Navigator возвращает ответ в формате json.
• Ответ включает:
  
  • поисковый запрос
  • количество результатов
  • список url гифок

Пример ответа:

{    "query": "Dependency Injector",    "limit": 10,    "gifs": [        {            "url": "https://giphy.com/gifs/boxes-dependent-swbf2-6Eo7KzABxgJMY"        },        {            "url": "https://giphy.com/gifs/depends-J56qCcOhk6hKE"        },        {            "url": "https://giphy.com/gifs/web-series-ccstudios-bro-dependent-1lhU8KAVwmVVu"        },        {            "url": "https://giphy.com/gifs/TheBoysTV-friends-friend-weneedeachother-XxR9qcIwcf5Jq404Sx"        },        {            "url": "https://giphy.com/gifs/netflix-a-series-of-unfortunate-events-asoue-9rgeQXbwoK53pcxn7f"        },        {            "url": "https://giphy.com/gifs/black-and-white-sad-skins-Hs4YzLs2zJuLu"        },        {            "url": "https://giphy.com/gifs/always-there-for-you-i-am-here-PlayjhCco9jHBYrd9w"        },        {            "url": "https://giphy.com/gifs/stream-famous-dollar-YT2dvOByEwXCdoYiA1"        },        {            "url": "https://giphy.com/gifs/i-love-you-there-for-am-1BhGzgpZXYWwWMAGB1"        },        {            "url": "https://giphy.com/gifs/life-like-twerk-9hlnWxjHqmH28"        }    ]}

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

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

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

mkdir giphynav-aiohttp-tutorialcd giphynav-aiohttp-tutorialpython3 -m venv venv

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

. venv/bin/activate

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

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

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

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

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

./ giphynavigator/    __init__.py    application.py    containers.py    views.py venv/ requirements.txt

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

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

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

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

dependency-injectoraiohttpaiohttp-devtoolspyyamlpytest-aiohttppytest-cov

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

pip install -r requirements.txt

Дополнительно установим httpie. Это HTTP клиент для командной строки. Мы будем
использовать его для ручного тестирования API.

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

pip install httpie

Зависимости установлены. Теперь построим минимальное приложение.

Минимальное приложение

В этом разделе построим минимальное приложение. У него будет эндпоинт, который будет возвращать пустой ответ.

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

"""Views module."""from aiohttp import webasync def index(request: web.Request) -> web.Response:    query = request.query.get('query', 'Dependency Injector')    limit = int(request.query.get('limit', 10))    gifs = []    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

Теперь добавим контейнер зависимостей (дальше просто контейнер). Контейнер будет содержать все компоненты приложения. Добавим первые два компонента. Это aiohttp приложение и представление index.

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

"""Application containers module."""from dependency_injector import containersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    index_view = aiohttp.View(views.index)

Теперь нам нужно создать фабрику aiohttp приложения. Ее обычно называют
create_app(). Она будет создавать контейнер. Контейнер будет использован для создания aiohttp приложения. Последним шагом настроим маршрутизацию мы назначим представление index_view из контейнера обрабатывать запросы к корню "/" нашего приложения.

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

"""Application module."""from aiohttp import webfrom .containers import ApplicationContainerdef create_app():    """Create and return aiohttp application."""    container = ApplicationContainer()    app: web.Application = container.app()    app.container = container    app.add_routes([        web.get('/', container.index_view.as_view()),    ])    return app

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

Теперь мы готовы запустить наше приложение:

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

adev runserver giphynavigator/application.py --livereload

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

[18:52:59] Starting aux server at http://localhost:8001 [18:52:59] Starting dev server at http://localhost:8000 

Используем httpie чтобы проверить работу сервера:

http http://127.0.0.1:8000/

Вы увидите:

HTTP/1.1 200 OKContent-Length: 844Content-Type: application/json; charset=utf-8Date: Wed, 29 Jul 2020 21:01:50 GMTServer: Python/3.8 aiohttp/3.6.2{    "gifs": [],    "limit": 10,    "query": "Dependency Injector"}

Минимальное приложение готово. Давайте подключим Giphy API.

Giphy API клиент

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

Создайте пустой файл giphy.py в пакете giphynavigator:

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    views.py venv/ requirements.txt

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

"""Giphy client module."""from aiohttp import ClientSession, ClientTimeoutclass GiphyClient:    API_URL = 'http://api.giphy.com/v1'    def __init__(self, api_key, timeout):        self._api_key = api_key        self._timeout = ClientTimeout(timeout)    async def search(self, query, limit):        """Make search API call and return result."""        if not query:            return []        url = f'{self.API_URL}/gifs/search'        params = {            'q': query,            'api_key': self._api_key,            'limit': limit,        }        async with ClientSession(timeout=self._timeout) as session:            async with session.get(url, params=params) as response:                if response.status != 200:                    response.raise_for_status()                return await response.json()

Теперь нам нужно добавить GiphyClient в контейнер. У GiphyClient есть две зависимости, которые нужно передать при его создании: API ключ и таймаут запроса. Для этого нам нужно будет воспользоваться двумя новыми провайдерами из модуля dependency_injector.providers:

• Провайдер Factory будет создавать GiphyClient.
• Провайдер Configuration будет передавать API ключ и таймаут GiphyClient.

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    index_view = aiohttp.View(views.index)

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

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

Теперь давайте добавим файл конфигурации.
Будем использовать YAML.

Создайте пустой файл config.yml в корне проекта:

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    views.py venv/ config.yml requirements.txt

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

giphy:  request_timeout: 10

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

Теперь нам нужно отредактировать create_app() чтобы сделать 2 действие при старте приложения:

• Загрузить конфигурацию из config.yml
• Загрузить API ключ из переменной окружения GIPHY_API_KEY

Отредактируйте application.py:

"""Application module."""from aiohttp import webfrom .containers import ApplicationContainerdef create_app():    """Create and return aiohttp application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.config.giphy.api_key.from_env('GIPHY_API_KEY')    app: web.Application = container.app()    app.container = container    app.add_routes([        web.get('/', container.index_view.as_view()),    ])    return app

Теперь нам нужно создать API ключ и установить его в переменную окружения.

Чтобы не тратить на это время сейчас используйте вот этот ключ:

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Для создания собственного ключа Giphy API следуйте этому руководству.

Создание Giphy API клиента и установка конфигурации завершена. Давайте перейдем к сервису поиска.

Сервис поиска

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

• Выполнять поиск
• Форматировать полученный ответ

SearchService будет использовать GiphyClient.

Создайте пустой файл services.py в пакете giphynavigator:

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    services.py    views.py venv/ requirements.txt

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

"""Services module."""from .giphy import GiphyClientclass SearchService:    def __init__(self, giphy_client: GiphyClient):        self._giphy_client = giphy_client    async def search(self, query, limit):        """Search for gifs and return formatted data."""        if not query:            return []        result = await self._giphy_client.search(query, limit)        return [{'url': gif['url']} for gif in result['data']]

При создании SearchService нужно передавать GiphyClient. Мы укажем это при добавлении SearchService в контейнер.

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(views.index)

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

Подключаем поиск

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

Отредактируйте views.py:

"""Views module."""from aiohttp import webfrom .services import SearchServiceasync def index(        request: web.Request,        search_service: SearchService,) -> web.Response:    query = request.query.get('query', 'Dependency Injector')    limit = int(request.query.get('limit', 10))    gifs = await search_service.search(query, limit)    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,    )

Убедитесь что приложение работает или выполните:

adev runserver giphynavigator/application.py --livereload

и сделайте запрос к API в терминале:

http http://localhost:8000/ query=="wow,it works" limit==5

Вы увидите:

HTTP/1.1 200 OKContent-Length: 850Content-Type: application/json; charset=utf-8Date: Wed, 29 Jul 2020 22:22:55 GMTServer: Python/3.8 aiohttp/3.6.2{    "gifs": [        {            "url": "https://giphy.com/gifs/discoverychannel-nugget-gold-rush-rick-ness-KGGPIlnC4hr4u2s3pY"        },        {            "url": "https://giphy.com/gifs/primevideoin-ll1hyBS2IrUPLE0E71"        },        {            "url": "https://giphy.com/gifs/jackman-works-jackmanworks-l4pTgQoCrmXq8Txlu"        },        {            "url": "https://giphy.com/gifs/cat-massage-at-work-l46CzMaOlJXAFuO3u"        },        {            "url": "https://giphy.com/gifs/everwhatproductions-fun-christmas-3oxHQCI8tKXoeW4IBq"        },    ],    "limit": 10,    "query": "wow,it works"}

Поиск работает.

Немного рефакторинга

Наше представление index содержит два hardcoded значения:

• Поисковый запрос по умолчанию
• Лимит количества результатов

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

Отредактируйте views.py:

"""Views module."""from aiohttp import webfrom .services import SearchServiceasync def index(        request: web.Request,        search_service: SearchService,        default_query: str,        default_limit: int,) -> web.Response:    query = request.query.get('query', default_query)    limit = int(request.query.get('limit', default_limit))    gifs = await search_service.search(query, limit)    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )

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

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

giphy:  request_timeout: 10search:  default_query: "Dependency Injector"  default_limit: 10

Рефакторинг закончен. Мы сделали наше приложение чище перенесли hardcoded значения в конфигурацию.

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

Добавляем тесты

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

Создайте пустой файл tests.py в пакете giphynavigator:

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    services.py    tests.py    views.py venv/ requirements.txt

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

"""Tests module."""from unittest import mockimport pytestfrom giphynavigator.application import create_appfrom giphynavigator.giphy import GiphyClient@pytest.fixturedef app():    return create_app()@pytest.fixturedef client(app, aiohttp_client, loop):    return loop.run_until_complete(aiohttp_client(app))async def test_index(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [            {'url': 'https://giphy.com/gif1.gif'},            {'url': 'https://giphy.com/gif2.gif'},        ],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get(            '/',            params={                'query': 'test',                'limit': 10,            },        )    assert response.status == 200    data = await response.json()    assert data == {        'query': 'test',        'limit': 10,        'gifs': [            {'url': 'https://giphy.com/gif1.gif'},            {'url': 'https://giphy.com/gif2.gif'},        ],    }async def test_index_no_data(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get('/')    assert response.status == 200    data = await response.json()    assert data['gifs'] == []async def test_index_default_params(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get('/')    assert response.status == 200    data = await response.json()    assert data['query'] == app.container.config.search.default_query()    assert data['limit'] == app.container.config.search.default_limit()

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

py.test giphynavigator/tests.py --cov=giphynavigator

Вы увидите:

platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0collected 3 itemsgiphynavigator/tests.py ...                                     [100%]---------- coverage: platform darwin, python 3.8.3-final-0 -----------Name                            Stmts   Miss  Cover---------------------------------------------------giphynavigator/__init__.py          0      0   100%giphynavigator/__main__.py          5      5     0%giphynavigator/application.py      10      0   100%giphynavigator/containers.py       10      0   100%giphynavigator/giphy.py            16     11    31%giphynavigator/services.py          9      1    89%giphynavigator/tests.py            35      0   100%giphynavigator/views.py             7      0   100%---------------------------------------------------TOTAL                              92     17    82%

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

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

Заключение

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

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )

Что дальше?

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