Русский
Русский
English
Статистика
Реклама

Inversion of control

Что можно положить в механизм Dependency Injection в Angular?

26.08.2020 16:11:24 | Автор: admin
Почти каждый разработчик на Angular может найти в Dependency Injection решение своей проблемы. Это хорошо было видно в комментариях к моей прошлой статье. Люди рассматривали различные варианты работы с данными из DI, сравнивали их удобство для той или иной ситуации. Это здорово, потому что такой простой инструмент дает нам столько возможностей.

Но несколько человек отписались мне, что им тяжело понять DI и его возможности в Angular. В интернете не так уж много материалов о том, как использовать DI эффективно, и для многих разработчиков он сводится к работе с глобальными сервисами или передачей глобальных данных из корня приложения в компоненты.

Давайте посмотрим на этот механизм в Angular чуть глубже.




Знаете ли вы свои зависимости?


Иногда нелегко понять, сколько зависимостей имеет ваш код.

Например, взгляните на этот псевдокласс и посчитайте, сколько зависимостей он имеет:

import { API_URL } from '../../../env/api-url';import { Logger } from '../../services/logger'; class PseudoClass {   request() {       fetch(API_URL).then(...);   }    onError(error) {       const logger = new Logger();        logger.log(document.location, error);   }}

Ответ
fetch браузерное API, опираемся на глобальную переменную, ожидая, что она объявлена.

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

new Logger() также импортированные данные из другого файла и пересоздания множества экземпляров класса, когда нам достаточно лишь одного.

document также браузерное API и завязка на глобальную переменную.


Ну и что же не так?


Например, такой класс тяжело протестировать, так как он зависит от импортированных данных из других файлов и конкретных сущностей в них.

Другая ситуация: document и fetch будут без проблем работать в вашем браузере. Но если однажды вам потребуется перенести приложение в Server Side Rendering, то в nodejs окружении необходимых глобальных переменных может не быть.

Так и что же за DI и зачем он нужен?


Механизм внедрения зависимостей управляет зависимостями внутри приложения. В принципе, для нас, как для Angular-разработчиков, эта система довольно простая. Есть две основные операции: положить что-то в дерево зависимостей или получить что-то из него.

Механизм внедрения зависимостей управляет зависимостями внутри приложения. В принципе, для нас, как для Angular-разработчиков, эта система довольно простая. Есть две основные операции: положить что-то в дерево зависимостей или получить что-то из него.

Если хотите рассмотреть DI с более теоретической стороны, почитайте о принципе инверсии управления. Также можете посмотреть интересные видеоматериалы по теме: серия видео про IoC и DI у Ильи Климова на русском или небольшое видео про IoC на английском.

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

Схема работы областей видимости в DI:


Что мы можем положить в DI?


Первая из операций с DI что-то положить в него. Собственно, для этого Angular позволяет нам прописывать providers-массив в декораторах наших модулей, компонентов или директив. Давайте посмотрим, из чего этот массив может состоять.

Providing класса


Обычно это знает каждый разработчик на Angular. Это тот случай, когда вы добавляете в приложение сервис.

Angular создает экземпляр класса, когда вы запрашиваете его в первый раз. А с Angular 6 мы можем и вовсе не прописывать классы в массив providers, а указать самому классу, в какое место в DI ему встать с providedIn:

providers: [   {       provide: SomeService,       useClass: SomeService   },   // Angular позволяет сократить эту запись как самый частый кейс:   SomeService]


Providing значения


Также через DI можно поставлять константные значения. Это может быть как простая строка с URL вашего API, так и сложный Observable с данными.

Providing значения обычно реализуется в связке с InjectionToken. Этот объект ключ для DI-механизма. Сначала вы говорите: Я хочу получить вот эти данные по такому ключу. А позже приходите к DI и спрашиваете: Есть ли что-то по этому ключу?

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

Лучше посмотреть это сразу в действии, поэтому давайте взглянем на stackblitz с примером:



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

  • Мы можем переопределить значение токена на любом уровне дерева DI, никак не меняя компоненты, которые его используют.
  • Мы можем мокировать значение токена подходящими данными при тестировании.
  • Компонент полностью изолирован и всегда будет работать одинаково, независимо от контекста.


Providing фабрики


На мой взгляд, это самый мощный инструмент в механизме внедрения зависимостей Angular.

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

Вот еще один stackbitz с подробным примером создания фабрики со стримом.



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

Providing существующего экземпляра


Не самый частый случай, но этот вариант бывает очень полезным инструментом.

Вы можете положить в токен сущность, которая уже была создана. Хорошо работает в связке с forwardRef.

Посмотрите еще один пример со stackblitz с директивой, которая имплементирует интерфейс и подменяет собой другой токен через useExisting. В этом примере мы хотим переопределить значение токена только в области видимости DI для дочерних компонентов элемента, на котором висит директива. Причем директива может быть любой главное, что она реализует необходимый интерфейс.



Хитрости с DI-декораторами


DI-декораторы позволяют сделать запросы к DI более гибкими.

Если вы не знаете все четыре декоратора, советую почитать вот эту статью на Медиуме. Статья на английском, но там очень классные и понятные визуализации по теме.

Не многие также знают, что DI-декораторы можно использовать в массиве deps, который готовит аргументы для фабрики в providers.

providers: [   {     provide: SOME_TOKEN,     /**      * Чтобы фабрика получила декорированное значение, используйте такой      * [new Decorator(), new Decorator(),..., TOKEN]      * синтаксис.      *      * В этом случае вы получите null, если не будет значения для      * OPTIONAL_TOKEN      */     deps: [[new Optional(), OPTIONAL_TOKEN]],     useFactory: someTokenFactory   } ]


Фабрика токена


Конструктор InjectionToken принимает два аргумента.

Второй аргумент объект с конфигурацией токена.

Фабрика токена это функция, которая вызывается в момент, когда кто-то запрашивает этот токен в первый раз. В ней можно посчитать некое стандартное значение для токена или даже обращаться к другим DI-сущностям через функцию inject.

Посмотрите пример реализации функционала стрима нажатия кнопок, но уже на фабрике токена.



Заключение


DI в Angular удивительная тема: на первый взгляд, в нем не так много различных рычагов и инструментов для изучения, но можно писать и говорить часами о тех возможностях и способах применения, которые они нам дают.

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

Domain-driven design, Hexagonal architecture of ports and adapters, Dependency injection и Python

31.05.2021 12:16:05 | Автор: admin

Prologue

- Глянь, статью на Хабр подготовил.
- Эм... а почему заголовок на английском?
- "Предметно-ориентированное проектирование, Гексагональная архитектура портов и адаптеров, Внедрение зависимостей и Пайто..."

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

Intro

Как же летит время! Два года назад я расстался с миром Django и очутился в мире Kotlin, Java и Spring Boot. Я испытал самый настоящий культурный шок. Голова гудела от объёма новых знаний. Хотелось бежать обратно в тёплую, ламповую, знакомую до байтов экосистему Питона. Особенно тяжело на первых порах давалась концепция инверсии управления (Inversion of Control, IoC) при связывании компонентов. После прямолинейного подхода Django, автоматическое внедрение зависимостей (Dependency Injection, DI) казалось чёрной магией. Но именно эта особенность фреймворка Spring Boot позволила проектировать приложения следуя заветам Чистой Архитектуры. Самым же большим вызовом стал отказ от философии "пилим фичи из трекера" в пользу Предметно-ориентированного проектирования (Domain-Driven Design, DDD).

Наш проект стремительно растёт и усложняется. Несмотря на это, его легко поддерживать, тестировать и развивать - всё благодаря качественному фундаменту и архитектуре. Код получается выразительным и легковесным. Компоненты - легко заменяемыми. Во всех отношениях это приложение качественнее всего написанного каждым членом нашей команды в прошлом.

Оглядываясь назад вспоминаю, какие пробелы в моём опыте и знаниях не позволяли писать и решать задачи бизнеса так элегантно. Если вы живёте в экоситеме Питона и на практике хотите познакомиться со всем перечисленным в заголовке, прошу!

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

Dependency Injection

Вы знаете что такое Внедрение зависимостей ака Dependency Injection (DI). Точно знаете, хотя можете и не вспомнить формального определения. Давайте на небольшом примере рассмотрим, в чём плюсы и минусы этого подхода (если вам угодно - шаблона).

Допустим нам понадобилась функция, отправляющая сообщения с пометкой "ТРЕВОГА!" в шину сообщений. После недолгих размышлений напишем:

from my_cool_messaging_library import get_message_bus()def send_alert(message: str):    message_bus = get_message_bus()    message_bus.send(topic='alert', message=message)

В чём главная проблема функции send_alert()? Она зависит от объекта message_bus, но для вызывающего эта зависимость совершенно не очевидна! А если вы хотите отправить сообщение по другой шине? А как насчёт уровня магии, необходимой для тестирования этой функции? Что, что? mock.patch(...) говорите? Коллеги, атака в лоб провалилась, давайте зайдём с флангов.

from my_cool_messaging_library import MessageBusdef send_alert(message_bus: MessageBus, message: str):    message_bus.send(topic='alert', message=message)

Казалось, небольшое изменение, добавили аргумент в функцию. Но одним лишь этим изменением мы убиваем нескольких зайцев: Вызывающему очевидно, что функция send_alert() зависит от объекта message_bus типа MessageBus (да здравствуют аннотации!). А тестирование, из обезьяньих патчей с бубном, превращается в написание краткого и ясного кода. Не верите?

def test_send_alert_sends_message_to_alert_topic()    message_bus_mock = MessageBusMock()    send_alert(message_bus_mock, "A week of astrology at Habrahabr!")    assert message_bus_mock.sent_to_topic == 'alert'    assert message_bus_mock.sent_message == "A week of astrology at Habrahabr!"class MessageBusMock(MessageBus):    def send(self, topic, message):        self.sent_to_topic = topic        self.sent_message = message

Тут искушённый читатель задастся вопросом: неужели придётся передавать экземпляр message_bus в функцию send_alert() при каждом вызове? Но ведь это неудобно! В чём смысл каждый раз писать

send_alert(get_message_bus(), "Stackoverflow is down")

Попытаемся решить эту проблему посредством ООП:

class AlertDispatcher:    _message_bus: MessageBus    def __init__(self, message_bus: MessageBus):        self._message_bus = message_bus    def send(message: str):        self._message_bus.send(topic='alert', message=message)alert_dispatcher = AlertDispatcher(get_message_bus())alert_dispatcher.send("Oh no, yet another dependency!")

Теперь уже класс AlertDispatcher зависит от объекта типа MessageBus. Мы внедряем эту зависимость в момент создания объекта AlertDispatcher посредством передачи зависимости в конструктор. Мы связали (we have wired, не путать с coupling!) объект и его зависимость.

Но теперь акцент смещается с message_bus на alert_dispatcher! Этот компонент может понадобиться в различных местах приложения. Мало ли откуда нужно оправить сигнал тревоги! Значит, необходим некий глобальный контекст из которого можно будет этот объект достать. И прежде чем перейти к построению такого контекста, давайте немного порассуждаем о природе компонентов и их связывании.

Componential architecture

Говоря о внедрении зависимостей мы не сильно заостряли внимание на типах. Но вы наверняка догадались, что MessageBus - это всего лишь абстракция, интерфейс, или как бы сказал PEP-544 - протокол. Где-то в нашем приложении объявленo:

class MessageBus(typing.Protocol):    def send(topic: str, message: str):        pass

В проекте также есть простейшая реализация MessageBus-a, записывающая сообщения в список:

class MemoryMessageBus(MessageBus):    sent_messages = []    def send(topic: str, messagge: str):        self.sent_messages.append((str, message))

Таким же образом можно абстрагировать бизнес-логику, разделив абстрактный сценарий пользования (use case) и его имплементацию:

class DispatchAlertUseCase(typing.Protocol):    def dispatch_alert(message: str):        pass
class AlertDispatcherService(DispatchAlertUseCase):    _message_bus: MessageBus    def __init__(self, message_bus: MessageBus):        self._message_bus = message_bus    def dispatch_alert(message: str):        self._message_bus.send(topic='alert', message=message)

Давайте для наглядности добавим HTTP-контроллер, который принимает сообщения по HTTP-каналу и вызывает DispatchAlertUseCase:

class ChatOpsController:    ...    def __init__(self, dispatch_alert_use_case: DispatchAlertUseCase):        self._dispatch_alert_use_case = dispatch_alert_use_case    @post('/alert)    def alert(self, message: Message):        self._dispatch_alert_use_case.dispatch_alert(message)        return HTTP_ACCEPTED

Наконец, всё это необходимо связать воедино:

from my_favourite_http_framework import http_serverdef main():    message_bus = MemoryMessageBus()    alert_dispatcher_service = AlertDispatcherService(message_bus)    chat_opts_controller = ChatOpsController(alert_dispatcher_service)    http_server.start()

Первой же реакцией здорового программиста будет: "ну нафига громоздить столько кода?". Ваша правда, всё вышенаписанное умещается в одну коротенькую функцию:

@post('/alert)def alert(message: Message):    bus = MemoryMessageBus()    bus.send(topic='alert', message=message)    return HTTP_ACCEPTED

Коротко? Ещё как! Поддерживаемо? Вообще никак. Почему? Из-за сильнейшей связанности (coupling) компонентов в коде. Уместив всё в одну функцию таким образом, мы намертво привязали логику отправки оповещений к конкретной реализации шины сообщений. Но это ещё полбеды. Самое ужасное то, что бизнес-составляющая полностью растворилась в технических деталях. Не поймите меня неправильно, подобный код вполне имеет право на существование. Но простит ли растущее приложение такой сжатый подход?

Вернёмся к нашей компонентной архитектуре. В чём её преимущества?

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

  • Каждый компонент работает в чётких рамках и решает лишь одну задачу.

  • Это значит, что компоненты могут быть протестированы как в полной изоляции, так и в любой произвольной комбинации включающей тестовых двойников (test double). Думаю не стоит объяснять, насколько проще тестировать изолированные части программы. Подход к TDD меняется с невнятного "нуууу, у нас есть тесты" на бодрое "тесты утром, вечером код".

  • С учётом того, что зависимости описываются абстракциями, можно безболезненно заменить один компонент другим. В нашем примере - вместо MemoryMessageBus можно бухнуть DbMessageBus, да хоть в файл на диске писать - тому кто вызывает message_bus.send(...) нет до этого никакого дела.

"Да это же SOLID!" - скажите вы. И будете абсолютно правы. Не удивлюсь, если у вас возникло чувство дежавю, ведь благородный дон @zueve год назад детально описал связь SOLID и Чистой архитектуры в статье "Clean Architecture глазами Python-разработчика". И наша компонентная архитектура находится лишь в шаге от чистой "гексагональной" архитектуры. Кстати, причём тут гексагон?

Architecture is about intent

Одно из замечательных высказываний дядюшки Боба на тему архитектуры приложений - Architecture is about intent (Намерения - в архитектуре).

Что вы видите на этом скриншоте?

Не удивлюсь, если многие ответили "Типичное приложение на Django". Отлично! А что же делает это приложение? Вы вероятно телепат 80го уровня, если смогли ответить на этот вопрос правильно. Лично я не именю ни малейшего понятия - это скриншот первого попавшегося Django-приложения с Гитхаба.

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

Разгадка

Это один из этажей библиотеки Oodi в Хельсинки.

Надеюсь вам было несложно отгадать эту маленькую загадку и вы вынесли из неё главное: архитектура должна встречать нас с порога, буквально с момента окончания git clone.... Как здорово, когда код приложения организован таким образом, что предназначение того или иного файла или директории лежит на поверхности!

В "Гексагональной архитектуре", гексагон в частности призван упростить восприятие архитектуры. Мудрено? Пардон, сейчас всё будет продемонстрировано наглядно.

Hexagonal architecture of Ports and Adapters

"У нас Гексагональная архитектура портов и адаптеров" - с этой фразы начинается рассказ об архитектуре приложения новым членам команды. Далее мы показываем нечто Ктулхуподобное:

Изобретатель термина "Гексагональная архитектура" Алистар Кокбёрн (Alistair Cockburn) объясняя выбор названия акцентировал внимание на его графическом представлении:

В гексагоне количество граней не играет роли. Важна сама возможность пририсовать порты и адаптеры по необходимости, выходя за рамки изображения послойных однонаправленных графов. Термин "гексагональная архитектура" вытекает из визуальной составляющей. (источник)

Итак, на изображении мы видим:

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

"Пользователь может голосовать за публикации, комментарии и карму других пользователей если его карма 5"

будет отображено именно здесь. И как вы наверняка поняли, в домене нет места HTTP, SQL, RabbitMQ, AWS и т.д. и т.п.

Зато всему этому празднику технологий есть место в адаптерах подсоединяемых к портам. Команды и запросы поступают в приложение через ведущие (driver) или API порты. Команды и запросы которые отдаёт приложение поступают в ведомые порты (driven port). Их также называют портами интерфейса поставщика услуг (Service Provider Interface, SPI).

Между портами и доменом сидят дирижёры - сервисы приложения (Application services). Они являются связующим звеном между сценариями пользования, доменом и ведомыми портами необходимыми для выполнения сценария. Также стоит упомянуть, что именно сервис приложения определяет, будет ли сценарий выполняться в рамках общей транзакции, или нет.

Всё это - и порты, и адаптеры и сервисы приложения и даже домен - слои архитектуры, состоящие из индивидуальных компонентов. Главной заповедью взаимодействия между слоями является "Зависимости всегда направлены от внешних слоёв к центру приложения". Например, адаптер может ссылаться на домен или другой адаптер, а домен ссылаться на адаптер - не может.

И... ВСЁ. Это - вся суть Гексагональной архитектуры портов и адаптеров. Она замечательно подходит для задач с обширной предметной областью. Для голого CRUDа а-ля HTTP интерфейс для базы данных, такая архитектура избыточна - Active Record вам в руки.

Давайте же засучим рукава и разберём на примере, как спроектировать Django-приложение по канонам гексагональной архитектуры.

Interlude

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

Во второй части вас ждёт реализация гексагональной архитектуры на знакомом нам всем примере. В первой части мы старались абстрагироваться от конкретных решений, будь то фреймворки или библиотеки. Последующий пример построен на основе Django и DRF с целью продемонстрировать, как можно вплести гексагональную архитектуру в фреймворк с устоявшимися традициями и архитектурными решениями. В приведённых примерах вырезаны некоторые необязательные участки и имеются допущения. Это сделано для того, чтобы мы могли сфокусироваться на важном и не отвлекались на второстепенные детали. Полностью исходный код примера доступен в репозитории https://github.com/basicWolf/hexagonal-architecture-django.

Upvote a post at Hubruhubr

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

Рейтинг публикации меняется путём голосования пользователей.

  1. Пользователь может проголосовать "ЗА" или "ПРОТИВ" публикации.

  2. Пользователь может голосовать если его карма 5.

  3. Проголосовать за данную публикацию можно лишь один раз, изменить голос нельзя.

С чего же начать работу? Конечно же с построения модели предметной области!

Domain model

Давайте ещё раз внимательно прочтём требования и подумаем, как описать "пользователя голосующего за публикацию"? Например (source):

# src/myapp/application/domain/model/voting_user.pyclass VotingUser:    id: UUID    voting_for_article_id: UUID    voted: bool    karma: int    def cast_vote(self, vote: Vote) -> CastArticleVoteResult:        ...

На первый взгляд - сомнительного вида творение. Но обратившись к деталям сценария мы убедимся, что этот набор данных - необходим и достаточен для голосования. Vote и CastArticleVoteResult - это также модели домена (source):

# src/myapp/application/domain/model/vote.py# Обозначает голос "За" или "Против"class Vote(Enum):    UP = 'up'    DOWN = 'down'

В свою очередь CastArticleVoteResult - это тип объединяющий оговорённые исходы сценария: ГолосПользователя, НедостаточноКармы, ПользовательУжеПроголосовалЗаПубликацию (source):

# src/myapp/application/domain/model/cast_article_vote_result.py...CastArticleVoteResult = Union[ArticleVote, InsufficientKarma, VoteAlreadyCast]

Как вы думаете, каких данных достаточно для описания результата успешно выполненного сценария?

Ответ

(source)

# src/myapp/application/domain/model/article_vote.py@dataclassclass ArticleVote:    user_id: UUID    article_id: UUID    vote: Vote    id: UUID = field(default_factory=uuid4)

Но самое интересное будет происходить в теле метода cast_article_vote(). И начнём мы конечно же с тестов. Первый же тест нацелен на проверку успешно выполненного сценария (source):

def test_cast_vote_returns_article_vote(user_id: UUID, article_id: UUID):    voting_user = VotingUser(        user_id=user_id,        voting_for_article_id=article_id,        karma=10    )    result = voting_user.cast_vote(Vote.UP)    assert isinstance(result, ArticleVote)    assert result.vote == Vote.UP    assert result.article_id == article_id    assert result.user_id == user_id

Запускаем тест и... ожидаемый фейл. В лучших традициях ТДД мы начнём игру в пинг-понг с тестами и кодом, с каждым тестом дописывая сценарий до полной готовности (source):

MINIMUM_KARMA_REQUIRED_FOR_VOTING = 5...def cast_vote(self, vote: Vote) -> CastArticleVoteResult1:    if self.voted:        return VoteAlreadyCast(            user_id=self.id,            article_id=self.voting_for_article_id        )    if self.karma < MINIMUM_KARMA_REQUIRED_FOR_VOTING:        return InsufficientKarma(user_id=self.id)    self.voted = True    return ArticleVote(        user_id=self.id,        article_id=self.voting_for_article_id,        vote=vote    )

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

Driver port: Cast article vote use case

Как было сказано ранее, в гексагональной архитектуре, приложение управляется через API-порты.

Чтобы как-то дотянуться до доменной модели, в наше приложение нужно добавить ведущий порт CastArticleVotingtUseCase, который принимает ID пользователя, ID публикации, значение голоса: за или против и возвращает результат выполненного сценария (source):

# src/myapp/application/ports/api/cast_article_vote/cast_aticle_vote_use_case.pyclass CastArticleVoteUseCase(Protocol):    def cast_article_vote(self, command: CastArticleVoteCommand) -> CastArticleVoteResult:        raise NotImplementedError()

Все входные параметры сценария обёрнуты в единую структуру-команду CastArticleVoteCommand (source), а все возможные результаты объединены - это уже знакомая модель домена CastArticleVoteResult (source):

# src/myapp/application/ports/api/cast_article_vote/cast_article_vote_command.py@dataclassclass CastArticleVoteCommand:    user_id: UUID    article_id: UUID    vote: Vote

Работа с гексагональной архитектурой чем-то напоминает прищурившегося Леонардо ди Каприо с фразой "We need to go deeper". Набросав каркас сценария пользования, можно примкнуть к нему с двух сторон. Можно имплементировать сервис, который свяжет доменную модель и ведомые порты для выполнения сценария. Или заняться API адаптерами, которые вызывают этот сценарий. Давайте зайдём со стороны API и напишем HTTP адаптер с помощью Django Rest Framework.

HTTP API Adapter

Наш HTTP адаптер, или на языке Django и DRF - View, до безобразия прост. За исключением преобразований запроса и ответа, он умещается в несколько строк (source):

# src/myapp/application/adapter/api/http/article_vote_view.pyclass ArticleVoteView(APIView):    ...    def __init__(self, cast_article_vote_use_case: CastArticleVoteUseCase):        self.cast_article_vote_use_case = cast_article_vote_use_case        super().__init__()    def post(self, request: Request) -> Response:        cast_article_vote_command = self._read_command(request)        result = self.cast_article_vote_use_case.cast_article_vote(            cast_article_vote_command        )        return self._build_response(result)    ...

И как вы поняли, смысл всего этого сводится к

  1. Принять HTTP запрос, десериализировать и валидировать входные данные.

  2. Запустить сценарий пользования.

  3. Сериализовать и возвратить результат выполненного сценария.

Этот адаптер конечно же строился по кирпичику с применением практик TDD и использованием инструментов Django и DRF для тестирования view-шек. Ведь для теста достаточно построить запрос (request), скормить его адаптеру и проверить ответ (response). При этом мы полностью контролируем основную зависимость cast_article_vote_use_case: CastArticleVoteUseCase и можем внедрить на её место тестового двойника.

Например, давайте напишем тест для сценария, в котором пользователь пытается проголосовать повторно. Ожидаемо, что статус в ответе будет 409 CONFLICT (source):

# tests/test_myapp/application/adapter/api/http/test_article_vote_view.pydef test_post_article_vote_with_same_user_and_article_id_twice_returns_conflict(    arf: APIRequestFactory,    user_id: UUID,    article_id: UUID):    # В роли объекта реализующего сценарий выступает    # специализированный двойник, возвращающий при вызове    # .cast_article_vote() контролируемый результат.    # Можно и MagicMock, но нужно ли?    cast_article_use_case_mock = CastArticleVoteUseCaseMock(        returned_result=VoteAlreadyCast(            user_id=user_id,            article_id=article_id        )    )    article_vote_view = ArticleVoteView.as_view(        cast_article_vote_use_case=cast_article_use_case_mock    )    response: Response = article_vote_view(        arf.post(            f'/article_vote',            {                'user_id': user_id,                'article_id': article_id,                'vote': Vote.UP.value            },            format='json'        )    )    assert response.status_code == HTTPStatus.CONFLICT    assert response.data == {        'status': 409,        'detail': f"User \"{user_id}\" has already cast a vote for article \"{article_id}\"",        'title': "Cannot cast a vote"    }

Адаптер получает на вход валидные данные, собирает из них команду и вызывает сценарий. Oднако, вместо продакшн-кода, этот вызов получает двойник, который тут же возвращает VoteAlreadyCast. Адаптеру же нужно правильно обработать этот результат и сформировать HTTP Response. Остаётся протестировать, соответствует ли сформированный ответ и его статус ожидаемым значениям.

Ещё раз попрошу заметить, насколько облегчённее становится тестирование, когда не нужно загружать всё приложение целиком. Адепты Django вспомнят о легковесном тестировании вьюшек посредством RequestFactory. Но гексагональная архитектура позволяет шагнуть дальше. Мы избавились от обезьяньих патчей и mock-обёрток конкретных классов. Мы легко управляем поведением зависимостей нашего View, ведь взаимодействие с ними происходит через абстрактный интерфейс. Всё это легко модифицировать и отлаживать.

После написания тестов и имплементации для остальных случаев входных и выходных данных, мы получаем отточенный API компонент. Следующим шагом нужно пристыковать этот компонент к рабочей версии сценария.

Application services

Как дирижёр управляет оркестром исполняющим произведение, так и сервис приложения управляет доменом и ведомыми портами при выполнении сценария.

PostRatingService

С места в карьер погрузимся в имплементацию нашего сценария. В первом приближении сервис реализующий сценарий выглядит так (source):

# src/myapp/application/service/post_rating_service.pyclass PostRatingService(    CastArticleVoteUseCase  # имплементируем протокол явным образом):    def cast_article_vote(self, command: CastArticleVoteCommand) -> CastArticleVoteResult:        ...

Отлично, но откуда возьмётся голосующий пользователь? Тут и появляется первая SPI-зависимость GetVotingUserPort задача которой найти голосующего пользователя по его ID. Но как мы помним, доменная модель не занимается записью голоса в какое-либо долговременное хранилище вроде БД. Для этого понадобится ещё одна SPI-зависимость SaveArticleVotePort:

# src/myapp/application/service/post_rating_service.pyclass PostRatingService(    CastArticleVoteUseCase):    _get_voting_user_port: GetVotingUserPort    _save_article_vote_port: SaveArticleVotePort    # def __init__(...) # внедрение зависимостей oпустим, чтобы не раздувать листинг    def cast_article_vote(self, command: CastArticleVoteCommand) -> CastArticleVoteResult:        voting_user = self._get_voting_user_port.get_voting_user(            user_id=command.user_id,            article_id=command.article_id        )        cast_vote_result = voting_user.cast_vote(command.vote)        if isinstance(cast_vote_result, ArticleVote):            self._save_article_vote_port.save_article_vote(cast_vote_result)        return cast_vote_result

Вы наверняка представили как выглядят интерфейсы этих SPI-зависимостей. Приведём один из интерфейсов здесь (source):

# src/myapp/application/ports/spi/save_article_vote_port.pyclass SaveArticleVotePort(Protocol):    def save_article_vote(self, article_vote: ArticleVote) -> ArticleVote:        raise NotImplementedError()

За кадром мы конечно же сначала напишем тесты, а уже потом код :) При написании юнит-тестов роль SPI-адаптеров в тестах сервиса, как и в предыдущих примерах, играют дублёры. Но чтобы удержать сей опус в рамках статьи, позвольте оставить тесты в виде ссылки на исходник (source) и двинуться дальше.

SPI Ports and Adapters

Продолжим рассматривать SPI-порты и адаптеры на примере SaveArticleVotePort. К этому моменту можно было и забыть, что мы всё ещё находимся в рамках Django. Ведь до сих пор не было написано того, с чего обычно начинается любое Django-приложение - модель данных! Начнём с адаптера, который можно подключить в вышеуказанный порт (source):

# src/myapp/application/adapter/spi/persistence/repository/article_vote_repository.pyfrom myapp.application.adapter.spi.persistence.entity.article_vote_entity import (    ArticleVoteEntity)from myapp.application.domain.model.article_vote import ArticleVotefrom myapp.application.ports.spi.save_article_vote_port import SaveArticleVotePortclass ArticleVoteRepository(    SaveArticleVotePort,):    def save_article_vote(self, article_vote: ArticleVote) -> ArticleVote:        article_vote_entity = ArticleVoteEntity.from_domain_model(article_vote)        article_vote_entity.save()        return article_vote_entity.to_domain_model()

Вспомним, что паттерн "Репозиторий" подразумевает скрытие деталей и тонкостей работы с источником данных. "Но позвольте! - скажете Вы, - a где здесь Django?". Чтобы избежать путаницы со словом "Model", модель данных носит гордое название ArticleVoteEntity. Entity также подразумевает, что у неё имеется уникальный идентификатор (source):

# src/myapp/application/adapter/spi/persistence/entity/article_vote_entity.pyclass ArticleVoteEntity(models.Model):    ... # здесь объявлены константы VOTE_UP, VOTE_DOWN и VOTE_CHOICES    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)    user_id = models.UUIDField()    article_id = models.UUIDField()    vote = models.IntegerField(choices=VOTES_CHOICES)    ...    def from_domain_model(cls, article_vote: ArticleVote) -> ArticleVoteEntity:        ...    def to_domain_model(self) -> ArticleVote:        ...

Таким образом, всё что происходит в save_article_vote() - это создание Django-модели из доменной модели, сохранение её в БД, обратная конвертация и возврат доменной модели. Это поведение легко протестировать. Например, юнит тест удачного исхода выглядит так (source):

# tests/test_myapp/application/adapter/spi/persistence/repository/test_article_vote_repository.py@pytest.mark.django_dbdef test_save_article_vote_persists_to_database(    article_vote_id: UUID,    user_id: UUID,    article_id: UUID):    article_vote_repository = ArticleVoteRepository()    article_vote_repository.save_article_vote(        ArticleVote(            id=article_vote_id,            user_id=user_id,            article_id=article_id,            vote=Vote.UP        )    )    assert ArticleVoteEntity.objects.filter(        id=article_vote_id,        user_id=user_id,        article_id=article_id,        vote=ArticleVoteEntity.VOTE_UP    ).exists()

Одним из требований Django является декларация моделей в models.py. Это решается простым импортированием:

# src/myapp/models.pyfrom myapp.application.adapter.spi.persistence.entity.article_vote_entity import ArticleVoteEntityfrom myapp.application.adapter.spi.persistence.entity.voting_user_entity import VotingUserEntity

Exceptions

Приложение почти готово!. Но вам не кажется, что мы кое-что упустили? Подсказка: Что произойдёт при голосовании, если ID пользователя или публикации будет указан неверно? Где-то в недрах Django вылетит исключение VotingUserEntity.DoesNotExist, что на поверхности выльется в неприятный HTTP 500 - Internal Server Error, хотя правильнее было бы вернуть HTTP 400 - Bad Request с телом, содержащим причину ошибки.

Ответ на вопрос, "В какой момент должно быть обработано это исключение?", вовсе не очевиден. С архитектурной точки зрения, ни API, ни домен не волнуют проблемы SPI-адаптеров. Максимум, что может сделать API с таким исключением - обработать его в общем порядке, а-ля except Exception:. С другой стороны SPI-порт может предоставить исключение-обёртку, в которую SPI-адаптер завернёт внутреннюю ошибку. А API может её поймать.

О, я слышу вас, дорогие адепты функционального программирования! "Какие исключения? В топку! Даёшь Either!". В ваших словах много правды и эта тема заслуживает отдельной статьи. В одном я же, я полностью соглашусь с вами - в домене не должно быть исключений!.

Например, в данной ситуации уместным будет исключение VotingUserNotFound (source) в которое оборачивается VotingUserEntity.DoesNotExist (source):

# src/myapp/application/adapter/spi/persistence/exceptions/voting_user_not_found.pyclass VotingUserNotFound(Exception):    def __init__(self, user_id: UUID):        super().__init__(user_id, f"User '{user_id}' not found")# ---# myapp/application/adapter/spi/persistence/repository/voting_user_repository.pyclass VotingUserRepository(GetVotingUserPort):    ...    def get_voting_user(self, user_id: UUID, article_id: UUID) -> VotingUser:        try:            # Код немного упрощён, в оригинале здесь происходит            # аннотация флагом "голосовал ли пользователь за статью".            # см. исходник            entity = VotingUserEntity.objects.get(id=user_id)        except VotingUserEntity.DoesNotExist as e:            raise VotingUserNotFound(user_id) from e        return self._to_domain_model(entity)

А вот теперь действительно, приложение почти готово! Осталось соединить все компоненты и точки входа.

Dependencies and application entry point

Традиционно точки входа и маршрутизация HTTP-запросов в Django-приложениях декларируется в urls.py. Всё что нам нужно сделать - это добавить запись в urlpatterns (source):

urlpatterns = [    path('article_vote', ArticleVoteView(...).as_view())]

Но погодите! Ведь ArticleVoteView требует зависимость имплементирующую CastArticleVoteUseCase. Это конечно же PostRatingService... которому в свою очередь требуются GetVotingUserPort и SaveArticleVotePort. Всю эту цепочку зависимостей удобно хранить и управлять из одного места - контейнера зависимостей (source):

# src/myapp/dependencies_container.py...def build_production_dependencies_container() -> Dict[str, Any]:    save_article_vote_adapter = ArticleVoteRepository()    get_vote_casting_user_adapter = VotingUserRepository()    cast_article_vote_use_case = PostRatingService(        get_vote_casting_user_adapter,        save_article_vote_adapter    )    article_vote_django_view = ArticleVoteView.as_view(        cast_article_vote_use_case=cast_article_vote_use_case    )    return {        'article_vote_django_view': article_vote_django_view    }

Этот контейнер инициализируется на старте приложения в AppConfig.ready() (source):

# myapp/apps.pyclass MyAppConfig(AppConfig):    name = 'myapp'    container: Dict[str, Any]    def ready(self) -> None:        from myapp.dependencies_container import build_production_dependencies_container        self.container = build_production_dependencies_container()

И наконец urls.py:

app_config = django_apps.get_containing_app_config('myapp')article_vote_django_view = app_config.container['article_vote_django_view']urlpatterns = [    path('article_vote', article_vote_django_view)]

Inversion of Control Containers

Для реализации одного небольшого сценария нам понадобилось создать и связать четыре компонента. С каждым новым сценарием, число компонентов будет расти и количество связей будет увеличиваться в арифметической прогрессии. Как управлять этим зоопарком, когда приложение начнёт разрастаться до неприличных размеров? Тут на помощь приходят Контейнеры Инверсии Управления.

IoC-container - это фреймворк управляющий объектами и их зависимостями во время исполнения программы.

Spring был первым универсальным IoC-контейнером / фреймворком с которым я столкнулся на практике (для зануд: Micronaut - да!). Чего уж таить, я не сразу проникся заложенными в него идеями. По-настоящему оценить всю мощь автоматического связывания (autowiring) и сопутствующего функционала я смог лишь выстраивая приложение следуя практикам гексагональной архитектуры.

Представьте, насколько удобнее будет использование условного декоратора @Component, который при загрузке программы внесёт класс в реестр зависимостей и выстроит дерево зависимостей автоматически?

T.e. если зарегистрировать компоненты:

@Componentclass ArticleVoteRepository(    SaveArticleVotePort,):    ...@Componentclass VotingUserRepository(GetVotingUserPort):    ...

То IoC-container сможет инициализировать и внедрить их через конструктор в другой компонент:

```@Componentclass PostRatingService(    CastArticleVoteUseCase):    def __init__(        self,        get_voting_user_port: GetVotingUserPort,        save_article_vote_port: SaveArticleVotePort    ):        ...

К сожалению мне не приходилось иметь дела с подобным инструментарием в экосистеме Питона. Буду благодарен, если вы поделитесь опытом в комментариях!

Directory structure

Помните скриншот "типичного Django-приложения"? Сравните его с тем что получилось у нас:

Чувствуете разницу? Нам больше не нужно лезть в файлы в надежде разобраться, что же там лежит и для чего они предназначены. Более того, теперь даже структура тестов и кода приложения идентичны! Архитектура приложения видна невооружённым глазом и существует "на бумаге", а не только в голове у разработчиков приложения.

Interlude

Давайте дружно выдохнем! Даю честное слово, больше ни одной строчки кода! Новый сценарий пользования готов к испытаниям. И пока коллеги вносят последние штрихи (миграция БД, отписки в трекере задач и т.п.) предлагаю поразмышлять вслух о том, почему гексагональная архитектура и предметно-ориентированное проектирование отлично подходят друг-другу.

Domain-Driven Design

Эрик Эванс (Eric Evans) популяризировал термин "Domain-Driven Design" в "большой синей книге" написанной в 2003м году. И всё заверте... Предметно-ориентированное проектирование - это методология разработки сложных систем, в которой во главу угла ставится понимание разработчиками предметной области путем общение с представителями (экспертами) предметной области и её моделирование в коде.

Мартин Фаулер (Martin Folwer) в своей статье рассуждая о заслугах Эванса подчёркивает, что в этой книге Эванс закрепил терминологию DDD, которой мы пользуемся и по сей день.

В частности, Эванс ввёл понятие об Универсальном Языке (Ubiquitous Language) - языке который разработчики с одной стороны и эксперты предметной области с другой, вырабатывают в процессе общения в течении всей жизни продукта. Невероятно сложно создать систему (а ведь смысл DDD - помочь нам проектировать именно сложные системы!) не понимая, для чего она предназначена и как ею пользуются.

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

- Майкл Крайтон, "Парк Юрского периода"

Более того, универсальный язык, со всеми оговорёнными терминами, сущностями, действиями, связями и т.д. используется при написании программы - в названиях модулей, функций, методов, классов, констант и даже переменных!

Другой важный термин - Ограниченный Контекст (Bounded Context) - автономные части предметной области с устоявшимися правилами, терминами и определениями. Простой пример: в онлайн магазине, модель "товар" несёт в себе совершенно разный смысл для отделов маркетинга, бухгалтерии, склада и логистики. Для связи моделей товара в этих контекстах достаточно наличие одинакового идентификатора (например UUID).

Понятие об Агрегатах (Aggregate) - наборе объектов предметной области, с которыми можно обращаться как единым целым, классификации объектов-значений и объектов-сущностей.

О DDD можно рассуждать и рассуждать. Эту тему не то что в одну статью, её и в толстенную книгу-то нелегко уместить. Приведу лишь несколько цитат, которые помогут перекинуть мостик между DDD и гексагональной архитектурой:

Предметная область - это сфера знаний или деятельности.

Модель - это система абстракций, представляющих определённый аспект предметной области.

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

Преимущество есть лишь у той модели, которая подходит для решения данной проблемы.

Эти цитаты взяты из выступления Эрика Эванса на конференции DDD Europe 2019 года. Приглашаю вас насладиться этим выступлением, прежде чем вы введёте "DDD" в поиск Хабра и начнёте увлекательное падение в бездонную кроличью нору. По пути вас ждёт много открытий и куча набитых шишек. Помню один восхитительный момент: внезапно в голове сложилась мозаика и пришло озарение, что фундаментальные идеи DDD и Agile Manifesto имеют общие корни.

Hexagonal Architecture

Так причём же здесь Гексагональная архитектура? Я очень надеюсь, что внимательный читатель уже ответил на этот вопрос.

На заре Гексагональной архитектуры в 2005м году, Алистар Кокбёрн писал:

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

Гексагональная архитектура позволяет элегантно изолировать части приложения и связать их посредством абстракций.

Становится просто связать модель предметной области в коде и "на бумаге" используя универсальный язык общения с экспертами. Универсальный язык обогащается с обеих сторон. При написании кода находятся и изменяются объекты, связи между ними и всё это перетекает обратно в модель предметной области.

Взаимодействие с внешним миром также упрощается, ведь оно происходит в рамках изолированных и взаимозаменяемых компонентов.

Тесты. Тэст-Дривэн Дэвэлопмэнт. В самом соке, когда тест пишется, к пока не существующему функционалу и мы даём возможность нашей IDE (или по-старинке) создать класс/метод/функцию/концепцию которая пока существует лишь в тесте. Интеграционные тесты, для которых необязательно загружать всю программу и инфраструктуру, а лишь адаптеры и необходимые для теста сервисы.

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

Microservices

Подумайте, каким образом в наши дни разбивают приложение на части, работа которых происходит в ограниченных контекстах? Ответ очевиден - Микросервисы! Все вышеописанные плюсы гексагональной архитектуры применимы и в этом случае. Но помимо вышеописанных плюсов, появляется возможность модифицировать микросервисы буквально методом "вырезать-вставить". Можно откреплять целые куски логики и кода из одних и вставлять в другие, отпочковывать микросервисы из монолита, или собирать всё обратно в монолит. И даже не важно, на каких языках написаны эти микросервисы. Ведь домен изолирован от технической составляющей! Поэтому переписывание на другой язык программирования становится куда более тривиальной задачей.

На десерт - короткое видео на тему от Дейва Фарли: The problem with microservices.

Outro

Спасибо вам уважаемый читатель. Спасибо, что не бросили меня в середине статьи и прошли со мной до конца. Надеюсь тема беседы вас заинтриговала и вы дерзнёте внедрить принципы гексагональной архитектуры и DDD в ваши проекты. Успехов и до новых встреч!

P.S.

Хотите проверить, насколько вы прониклись вышеизложенным? Тогда подумайте и ответьте, является ли поле VotingUser.voted оптимальным решением с точки зрения моделирования предметной области? А если нет, что бы вы предложили взамен?

Подробнее..

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

25.07.2020 06:07:48 | Автор: admin
Привет,

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

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

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

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

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

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

  • Python 3.5+
  • Virtual environment

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

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

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


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

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

  • Пользователь открывает веб-страницу где ему предлагают ввести поисковый запрос.
  • Пользователь вводит запрос и нажимает Enter.
  • Github Navigator ищет подходящие репозитории на Github.
  • По окончанию поиска Github Navigator показывает пользователю веб-страницу с результатами.
  • Страница результатов показывает все найденные репозитории и поисковый запрос.
  • Для каждого репозитория пользователь видит:
    • имя репозитория
    • владельца репозитория
    • последний коммит в репозиторий
  • Пользователь может нажать на любой из элементов чтобы открыть его страницу на Github.



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


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

mkdir ghnav-flask-tutorialcd ghnav-flask-tutorialpython3 -m venv venv

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

. venv/bin/activate

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

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


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

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

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

Пришло время установить Flask и Dependency Injector.

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

dependency-injectorflask

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

pip install -r requirements.txt

И проверим что установка прошла успешно:

python -c "import dependency_injector; print(dependency_injector.__version__)"python -c "import flask; print(flask.__version__)"

Вы увидите что-то вроде:

(venv) $ python -c "import dependency_injector; print(dependency_injector.__version__)"3.22.0(venv) $ python -c "import flask; print(flask.__version__)"1.1.2

Hello world!


Давайте создадим минимальное hello world приложение.

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

"""Views module."""def index():    return 'Hello, World!'

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

Добавим следующее в файл containers.py:

"""Application containers module."""from dependency_injector import containersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom . import viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    index_view = flask.View(views.index)

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

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

"""Application module."""from .containers import ApplicationContainerdef create_app():    """Create and return Flask application."""    container = ApplicationContainer()    app = container.app()    app.container = container    app.add_url_rule('/', view_func=container.index_view.as_view())    return app

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

Теперь наше приложение готово сказать Hello, World!.

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

export FLASK_APP=githubnavigator.applicationexport FLASK_ENV=developmentflask run

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

* Serving Flask app "githubnavigator.application" (lazy loading)* Environment: development* Debug mode: on* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)* Restarting with fsevents reloader* Debugger is active!* Debugger PIN: 473-587-859

Откройте браузер и зайдите на http://127.0.0.1:5000/.

Вы увидите Hello, World!.

Отлично. Наше минимальное приложение успешно стартует и работает.

Давайте сделаем его немного красивее.

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


Мы будем использовать Bootstrap 4. Используем для этого расширение Bootstrap-Flask. Оно поможет нам добавить все нужные файлы в несколько кликов.

Добавим bootstrap-flask в requirements.txt:

dependency-injectorflaskbootstrap-flask

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

pip install --upgrade -r requirements.txt

Теперь добавим расширение bootstrap-flask в контейнер.

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

"""Application containers module."""from dependency_injector import containersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom . import viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    index_view = flask.View(views.index)

Давайте инициализируем расширение bootstrap-flask. Нам нужно будет изменить create_app().

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

"""Application module."""from .containers import ApplicationContainerdef create_app():    """Create and return Flask application."""    container = ApplicationContainer()    app = container.app()    app.container = container    bootstrap = container.bootstrap()    bootstrap.init_app(app)    app.add_url_rule('/', view_func=container.index_view.as_view())    return app

Теперь нужно добавить шаблоны. Для этого нам понадобится добавить папку templates/ в пакет githubnavigator. Внутри папки с шаблонами добавим два файла:

  • base.html базовый шаблон
  • index.html шаблон основной страницы

Создаем папку templates и два пустых файла внутри base.html и index.html:

./ githubnavigator/    templates/       base.html       index.html    __init__.py    application.py    containers.py    views.py venv/ requirements.txt

Теперь давайте наполним базовый шаблон.

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

<!doctype html><html lang="en">    <head>        {% block head %}        <!-- Required meta tags -->        <meta charset="utf-8">        <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">        {% block styles %}            <!-- Bootstrap CSS -->            {{ bootstrap.load_css() }}        {% endblock %}        <title>{% block title %}{% endblock %}</title>        {% endblock %}    </head>    <body>        <!-- Your page content -->        {% block content %}{% endblock %}        {% block scripts %}            <!-- Optional JavaScript -->            {{ bootstrap.load_js() }}        {% endblock %}    </body></html>

Теперь наполним шаблон основной страницы.

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

{% extends "base.html" %}{% block title %}Github Navigator{% endblock %}{% block content %}<div class="container">    <h1 class="mb-4">Github Navigator</h1>    <form>        <div class="form-group form-row">            <div class="col-10">                <label for="search_query" class="col-form-label">                    Search for:                </label>                <input class="form-control" type="text" id="search_query"                       placeholder="Type something to search on the GitHub"                       name="query"                       value="{{ query if query }}">            </div>            <div class="col">                <label for="search_limit" class="col-form-label">                    Limit:                </label>                <select class="form-control" id="search_limit" name="limit">                    {% for value in [5, 10, 20] %}                    <option {% if value == limit %}selected{% endif %}>                        {{ value }}                    </option>                    {% endfor %}                </select>            </div>        </div>    </form>    <p><small>Results found: {{ repositories|length }}</small></p>    <table class="table table-striped">        <thead>            <tr>                <th>#</th>                <th>Repository</th>                <th class="text-nowrap">Repository owner</th>                <th class="text-nowrap">Last commit</th>            </tr>        </thead>        <tbody>        {% for repository in repositories %} {{n}}            <tr>              <th>{{ loop.index }}</th>              <td><a href="{{ repository.url }}">                  {{ repository.name }}</a>              </td>              <td><a href="{{ repository.owner.url }}">                  <img src="{{ repository.owner.avatar_url }}"                       alt="avatar" height="24" width="24"/></a>                  <a href="{{ repository.owner.url }}">                      {{ repository.owner.login }}</a>              </td>              <td><a href="{{ repository.latest_commit.url }}">                  {{ repository.latest_commit.sha }}</a>                  {{ repository.latest_commit.message }}                  {{ repository.latest_commit.author_name }}              </td>            </tr>        {% endfor %}        </tbody>    </table></div>{% endblock %}

Отлично, почти готово. Последним шагом изменим представление index чтобы оно использовало шаблон index.html.

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

"""Views module."""from flask import request, render_templatedef index():    query = request.args.get('query', 'Dependency Injector')    limit = request.args.get('limit', 10, int)    repositories = []    return render_template(        'index.html',        query=query,        limit=limit,        repositories=repositories,    )

Готово.

Убедитесь что приложение работает или выполните flask run и откройте http://127.0.0.1:5000/.

Вы должны увидите:



Подключаем Github


В этом разделе интегрируем наше приложение с Github API.
Мы будем использовать библиотеку PyGithub.

Добавим её в requirements.txt:

dependency-injectorflaskbootstrap-flaskpygithub

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

pip install --upgrade -r requirements.txt

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

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

Сделаем это.

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom github import Githubfrom . import viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    config = providers.Configuration()    github_client = providers.Factory(        Github,        login_or_token=config.github.auth_token,        timeout=config.github.request_timeout,    )    index_view = flask.View(views.index)

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

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

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

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

./ githubnavigator/    templates/       base.html       index.html    __init__.py    application.py    containers.py    views.py venv/ config.yml requirements.txt

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

github:  request_timeout: 10

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

Отредактируйте requirements.txt:

dependency-injectorflaskbootstrap-flaskpygithubpyyaml

и установите зависимость:

pip install --upgrade -r requirements.txt

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

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

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

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

"""Application module."""from .containers import ApplicationContainerdef create_app():    """Create and return Flask application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.config.github.auth_token.from_env('GITHUB_TOKEN')    app = container.app()    app.container = container    bootstrap = container.bootstrap()    bootstrap.init_app(app)    app.add_url_rule('/', view_func=container.index_view.as_view())    return app

Теперь нам нужно создать API токен.

Для это нужно:

  • Следовать этому руководству на Github
  • Установить токен в переменную окружения:

    export GITHUB_TOKEN=<your token>
    

Этот пункт можно временно пропустить.

Приложение будет работать без токена, но с ограниченной пропускной способностью. Ограничение для неаутентифицированных клиентов: 60 запросов в час. Токен нужен чтобы увеличить эту квоту до 5000 в час.

Готово.

Установка Github API клиента завершена.

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


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

  • Выполнять поиск на Github
  • Получать дополнительные данные о коммитах
  • Преобразовывать формат результат

SearchService будет использовать Github API клиент.

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

./ githubnavigator/    templates/       base.html       index.html    __init__.py    application.py    containers.py    services.py    views.py venv/ config.yml requirements.txt

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

"""Services module."""from github import Githubfrom github.Repository import Repositoryfrom github.Commit import Commitclass SearchService:    """Search service performs search on Github."""    def __init__(self, github_client: Github):        self._github_client = github_client    def search_repositories(self, query, limit):        """Search for repositories and return formatted data."""        repositories = self._github_client.search_repositories(            query=query,            **{'in': 'name'},        )        return [            self._format_repo(repository)            for repository in repositories[:limit]        ]    def _format_repo(self, repository: Repository):        commits = repository.get_commits()        return {            'url': repository.html_url,            'name': repository.name,            'owner': {                'login': repository.owner.login,                'url': repository.owner.html_url,                'avatar_url': repository.owner.avatar_url,            },            'latest_commit': self._format_commit(commits[0]) if commits else {},        }    def _format_commit(self, commit: Commit):        return {            'sha': commit.sha,            'url': commit.html_url,            'message': commit.commit.message,            'author_name': commit.commit.author.name,        }

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom github import Githubfrom . import services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    config = providers.Configuration()    github_client = providers.Factory(        Github,        login_or_token=config.github.auth_token,        timeout=config.github.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        github_client=github_client,    )    index_view = flask.View(views.index)

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


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

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

"""Views module."""from flask import request, render_templatefrom .services import SearchServicedef index(search_service: SearchService):    query = request.args.get('query', 'Dependency Injector')    limit = request.args.get('limit', 10, int)    repositories = search_service.search_repositories(query, limit)    return render_template(        'index.html',        query=query,        limit=limit,        repositories=repositories,    )

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom github import Githubfrom . import services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    config = providers.Configuration()    github_client = providers.Factory(        Github,        login_or_token=config.github.auth_token,        timeout=config.github.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        github_client=github_client,    )    index_view = flask.View(        views.index,        search_service=search_service,    )

Убедитесь что приложение работает или выполните flask run и откройте http://127.0.0.1:5000/.

Вы увидите:



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


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

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

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

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

"""Views module."""from flask import request, render_templatefrom .services import SearchServicedef index(        search_service: SearchService,        default_query: str,        default_limit: int,):    query = request.args.get('query', default_query)    limit = request.args.get('limit', default_limit, int)    repositories = search_service.search_repositories(query, limit)    return render_template(        'index.html',        query=query,        limit=limit,        repositories=repositories,    )

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

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

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom github import Githubfrom . import services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    config = providers.Configuration()    github_client = providers.Factory(        Github,        login_or_token=config.github.auth_token,        timeout=config.github.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        github_client=github_client,    )    index_view = flask.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )

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

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

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

Готово.

Рефакторинг закончен. Му сделали код чище.

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


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

Мы будем использовать pytest и coverage.

Отредактируйте requirements.txt:

dependency-injectorflaskbootstrap-flaskpygithubpyyamlpytest-flaskpytest-cov

и установите новые пакеты:

pip install -r requirements.txt

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

./ githubnavigator/    templates/       base.html       index.html    __init__.py    application.py    containers.py    services.py    tests.py    views.py venv/ config.yml requirements.txt

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

"""Tests module."""from unittest import mockimport pytestfrom github import Githubfrom flask import url_forfrom .application import create_app@pytest.fixturedef app():    return create_app()def test_index(client, app):    github_client_mock = mock.Mock(spec=Github)    github_client_mock.search_repositories.return_value = [        mock.Mock(            html_url='repo1-url',            name='repo1-name',            owner=mock.Mock(                login='owner1-login',                html_url='owner1-url',                avatar_url='owner1-avatar-url',            ),            get_commits=mock.Mock(return_value=[mock.Mock()]),        ),        mock.Mock(            html_url='repo2-url',            name='repo2-name',            owner=mock.Mock(                login='owner2-login',                html_url='owner2-url',                avatar_url='owner2-avatar-url',            ),            get_commits=mock.Mock(return_value=[mock.Mock()]),        ),    ]    with app.container.github_client.override(github_client_mock):        response = client.get(url_for('index'))    assert response.status_code == 200    assert b'Results found: 2' in response.data    assert b'repo1-url' in response.data    assert b'repo1-name' in response.data    assert b'owner1-login' in response.data    assert b'owner1-url' in response.data    assert b'owner1-avatar-url' in response.data    assert b'repo2-url' in response.data    assert b'repo2-name' in response.data    assert b'owner2-login' in response.data    assert b'owner2-url' in response.data    assert b'owner2-avatar-url' in response.datadef test_index_no_results(client, app):    github_client_mock = mock.Mock(spec=Github)    github_client_mock.search_repositories.return_value = []    with app.container.github_client.override(github_client_mock):        response = client.get(url_for('index'))    assert response.status_code == 200    assert b'Results found: 0' in response.data

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

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

Вы увидите:

platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1plugins: flask-1.0.0, cov-2.10.0collected 2 itemsgithubnavigator/tests.py ..                                     [100%]---------- coverage: platform darwin, python 3.8.3-final-0 -----------Name                             Stmts   Miss  Cover----------------------------------------------------githubnavigator/__init__.py          0      0   100%githubnavigator/application.py      11      0   100%githubnavigator/containers.py       13      0   100%githubnavigator/services.py         14      0   100%githubnavigator/tests.py            32      0   100%githubnavigator/views.py             7      0   100%----------------------------------------------------TOTAL                               77      0   100%

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

Заключение


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

Основная часть нашего приложения это контейнер. Он содержит все компоненты приложения и их зависимости в одном месте. Это предоставляет контроль над структурой приложения. Её легко понимать и изменять:

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import flaskfrom flask import Flaskfrom flask_bootstrap import Bootstrapfrom github import Githubfrom . import services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = flask.Application(Flask, __name__)    bootstrap = flask.Extension(Bootstrap)    config = providers.Configuration()    github_client = providers.Factory(        Github,        login_or_token=config.github.auth_token,        timeout=config.github.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        github_client=github_client,    )    index_view = flask.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )


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

Что дальше?


Подробнее..

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

02.08.2020 22:16:32 | Автор: admin
Привет,

Я создатель 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,    )

Что дальше?


Подробнее..

Мониторинг демон на Asyncio Dependency Injector руководство по применению dependency injection

09.08.2020 08:06:15 | Автор: admin
Привет,

Я создатель 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 не установлены, их нужно установить перед тем как продолжить. Следуйте этим руководствам:


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

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


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

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,        ),    )


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

Что дальше?


Подробнее..

CLI приложение Dependency Injector руководство по применению dependency injection Вопросы ответы

14.08.2020 02:18:06 | Автор: admin
Привет,

Я создатель 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+ компонентов в контейнере. Если не понравится много не потеряйте. Если понравится получите существенное преимущество.


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

Категории

Последние комментарии

  • Имя: Макс
    24.08.2022 | 11:28
    Я разраб в IT компании, работаю на арбитражную команду. Мы работаем с приламы и сайтами, при работе замечаются постоянные баны и лаги. Пацаны посоветовали сервис по анализу исходного кода,https://app Подробнее..
  • Имя: 9055410337
    20.08.2022 | 17:41
    поможем пишите в телеграм Подробнее..
  • Имя: sabbat
    17.08.2022 | 20:42
    Охренеть.. это просто шикарная статья, феноменально круто. Большое спасибо за разбор! Надеюсь как-нибудь с тобой связаться для обсуждений чего-либо) Подробнее..
  • Имя: Мария
    09.08.2022 | 14:44
    Добрый день. Если обладаете такой информацией, то подскажите, пожалуйста, где можно найти много-много материала по Yggdrasil и его уязвимостях для написания диплома? Благодарю. Подробнее..
© 2006-2024, personeltest.ru