[Python, Проектирование и рефакторинг] Aiohttp + Dependency Injector — руководство по применению dependency injection
Автор
Сообщение
news_bot ®
Стаж: 6 лет 9 месяцев
Сообщений: 27286
Привет,
Я создатель Dependency Injector. Это dependency injection фреймворк для Python.
Продолжаю серию руководств по применению Dependency Injector для построения приложений.
В этом руководстве хочу показать как применять Dependency Injector для разработки aiohttp приложений.
Руководство состоит из таких частей:
- Что мы будем строить?
- Подготовка окружения
- Структура проекта
- Установка зависимостей
- Минимальное приложение
- Giphy API клиент
- Сервис поиска
- Подключаем поиск
- Немного рефакторинга
- Добавляем тесты
- Заключение
Завершенный проект можно найти на 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-tutorial
cd giphynav-aiohttp-tutorial
python3 -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-injector
aiohttp
aiohttp-devtools
pyyaml
pytest-aiohttp
pytest-cov
И выполним в терминале:
pip install -r requirements.txt
Дополнительно установим httpie. Это HTTP клиент для командной строки. Мы будем
использовать его для ручного тестирования API.
Выполним в терминале:
pip install httpie
Зависимости установлены. Теперь построим минимальное приложение.
Минимальное приложение
В этом разделе построим минимальное приложение. У него будет эндпоинт, который будет возвращать пустой ответ.
Отредактируем views.py:
"""Views module."""
from aiohttp import web
async 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 containers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import views
class 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 web
from .containers import ApplicationContainer
def 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 OK
Content-Length: 844
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 21:01:50 GMT
Server: 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, ClientTimeout
class 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, providers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import giphy, views
class 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 web
from .containers import ApplicationContainer
def 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 GiphyClient
class 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, providers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import giphy, services, views
class 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 web
from .services import SearchService
async 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, providers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import giphy, services, views
class 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 OK
Content-Length: 850
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 22:22:55 GMT
Server: 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 web
from .services import SearchService
async 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, providers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import giphy, services, views
class 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: 10
search:
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 mock
import pytest
from giphynavigator.application import create_app
from giphynavigator.giphy import GiphyClient
@pytest.fixture
def app():
return create_app()
@pytest.fixture
def 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.1
plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0
collected 3 items
giphynavigator/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, providers
from dependency_injector.ext import aiohttp
from aiohttp import web
from . import giphy, services, views
class 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
===========
Источник:
habr.com
===========
Похожие новости:
- [Python, Будущее здесь, ООП, Параллельное программирование] Мир без корутин. Костыли для программиста — asyncio
- [Java, Программирование, Проектирование и рефакторинг, Разработка под Android, Совершенный код] Руководство Google по форматированию кода на Java (перевод)
- [Python, Системы сборки, Тестирование веб-сервисов] Тесты на pytest с генерацией отчетов в Allure с использованием Docker и Gitlab Pages и частично selenium
- Обновление JPype 1.0.2, библиотеки для доступа к Java-классам из Python
- [GitHub, Python, SQLite, Алгоритмы, Веб-аналитика] Как проанализировать рынок фотостудий с помощью Python (2/3). База данных
- [Разработка веб-сайтов] День и ночь в интернете, или открытое письмо веб-разработчикам
- [Проектирование и рефакторинг, Управление персоналом, Управление проектами] Как делать в два раза больше и получать от этого удовольствие
- [Разработка веб-сайтов, Python, Программирование, Функциональное программирование] Какая асинхронность должна была бы быть в Python
- [PostgreSQL, SQL, Администрирование баз данных] Пишем и тестируем миграции БД с Alembic. Доклад Яндекса
- [API, Python, Контекстная реклама, Яндекс API] Обзор python-пакета yadirstat — самый простой способ получить статистику из API Яндекс Директ
Теги для поиска: #_python, #_proektirovanie_i_refaktoring (Проектирование и рефакторинг), #_python, #_python3, #_aiohttp, #_dependency_injection, #_inversion_of_control, #_refactoring, #_object_oriented_design, #_python, #_proektirovanie_i_refaktoring (
Проектирование и рефакторинг
)
Вы не можете начинать темы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Текущее время: 23-Ноя 04:17
Часовой пояс: UTC + 5
| Автор | Сообщение |
|---|---|
|
news_bot ®
Стаж: 6 лет 9 месяцев |
|
|
Привет, Я создатель Dependency Injector. Это dependency injection фреймворк для Python. Продолжаю серию руководств по применению Dependency Injector для построения приложений. В этом руководстве хочу показать как применять Dependency Injector для разработки aiohttp приложений. Руководство состоит из таких частей:
Завершенный проект можно найти на Github. Для старта необходимо иметь:
И желательно иметь:
Что мы будем строить?  Мы будем строить REST API приложение, которое ищет забавные гифки на Giphy. Назовем его Giphy Navigator. Как работает Giphy Navigator?
Пример ответа: {
"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-tutorial
cd giphynav-aiohttp-tutorial python3 -m venv venv Теперь давайте активируем virtual environment: . venv/bin/activate
Окружение готово, теперь займемся структурой проекта. Структура проекта В этом разделе организуем структуру проекта. Создадим в текущей папке следующую структуру. Все файлы пока оставляем пустыми. Начальная структура: ./
├── giphynavigator/ │ ├── __init__.py │ ├── application.py │ ├── containers.py │ └── views.py ├── venv/ └── requirements.txt Установка зависимостей Пришло время установить зависимости. Мы будем использовать такие пакеты:
Добавим следующие строки в файл requirements.txt: dependency-injector
aiohttp aiohttp-devtools pyyaml pytest-aiohttp pytest-cov И выполним в терминале: pip install -r requirements.txt
Дополнительно установим httpie. Это HTTP клиент для командной строки. Мы будем использовать его для ручного тестирования API. Выполним в терминале: pip install httpie
Зависимости установлены. Теперь построим минимальное приложение. Минимальное приложение В этом разделе построим минимальное приложение. У него будет эндпоинт, который будет возвращать пустой ответ. Отредактируем views.py: """Views module."""
from aiohttp import web async 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 containers from dependency_injector.ext import aiohttp from aiohttp import web from . import views class 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 web from .containers import ApplicationContainer def 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 OK
Content-Length: 844 Content-Type: application/json; charset=utf-8 Date: Wed, 29 Jul 2020 21:01:50 GMT Server: 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, ClientTimeout class 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:
Отредактируем containers.py: """Application containers module."""
from dependency_injector import containers, providers from dependency_injector.ext import aiohttp from aiohttp import web from . import giphy, views class 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 действие при старте приложения:
Отредактируйте application.py: """Application module."""
from aiohttp import web from .containers import ApplicationContainer def 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 GiphyClient class 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, providers from dependency_injector.ext import aiohttp from aiohttp import web from . import giphy, services, views class 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 web from .services import SearchService async 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, providers from dependency_injector.ext import aiohttp from aiohttp import web from . import giphy, services, views class 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 OK
Content-Length: 850 Content-Type: application/json; charset=utf-8 Date: Wed, 29 Jul 2020 22:22:55 GMT Server: 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 web from .services import SearchService async 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, providers from dependency_injector.ext import aiohttp from aiohttp import web from . import giphy, services, views class 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: 10 search: 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 mock import pytest from giphynavigator.application import create_app from giphynavigator.giphy import GiphyClient @pytest.fixture def app(): return create_app() @pytest.fixture def 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.1
plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0 collected 3 items giphynavigator/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, providers from dependency_injector.ext import aiohttp from aiohttp import web from . import giphy, services, views class 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, ) Что дальше?
=========== Источник: habr.com =========== Похожие новости:
Проектирование и рефакторинг ) |
|
Вы не можете начинать темы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Вы не можете отвечать на сообщения
Вы не можете редактировать свои сообщения
Вы не можете удалять свои сообщения
Вы не можете голосовать в опросах
Вы не можете прикреплять файлы к сообщениям
Вы не можете скачивать файлы
Текущее время: 23-Ноя 04:17
Часовой пояс: UTC + 5