Внедрение зависимостей проще простого – на Python

Зачем нам требуется внедрение зависимостей?

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

from typing import List, Optional


class UserMessageSource:
    def get_user_message(self) -> str:
        raise NotImplementedError


class OutputWriter:
    def write_bot_messages(self, bot_messages: List[str]) -> None:
        raise NotImplementedError


class AnswerGenerator:
    def __init__(self):
        self.end_conversation = False

    def get_answers(self, user_message: str) -> List[str]:
        bot_messages = []
        if user_message in ["hello", "hi"]:
            bot_messages.append("Hello there!")
        elif user_message in ["bye", "good bye"]:
            bot_messages.append("See you!")
            self.end_conversation = True
        else:
            bot_messages.append("I'm sorry, I didn't understand that :(")
        return bot_messages


class ConversationLogger:
    def __init__(self, file_path: str):
        self.file_path = file_path

    def append_to_conversation(self, user_message: str, bot_messages: List[str]) -> None:
        with open(self.file_path, "a") as conversation_file:
            conversation_file.write(f"Human: {user_message}\n")
            for message in bot_messages:
                conversation_file.write(f"Bot: {message}\n")


class Chat:
    def __init__(self,
                 user_message_source: UserMessageSource,
                 output_writer: OutputWriter,
                 answer_generator: AnswerGenerator,
                 conversation_logger: Optional[ConversationLogger] = None):
        self.user_message_source = user_message_source
        self.output_writer = output_writer
        self.answer_generator = answer_generator
        self.conversation_logger = conversation_logger

    def run(self):
        while not self.answer_generator.end_conversation:
            user_message = self.user_message_source.get_user_message()
            bot_messages = self.answer_generator.get_answers(user_message)
            self.output_writer.write_bot_messages(bot_messages)
            if self.conversation_logger:
                self.conversation_logger.append_to_conversation(user_message, bot_messages)

Простое приложение для чата

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

UserMessageSource и OutputMessageWriter – это абстрактные классы, которые можно реализовать для кастомизации поведения чатбота.

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

from typing import List

from .chat import AnswerGenerator, Chat, ConversationLogger, OutputWriter, UserMessageSource


class CliUserMessageSource(UserMessageSource):
    def get_user_message(self) -> str:
        return input("Human: ").strip().lower()


class CliOutputWriter(OutputWriter):
    def write_bot_messages(self, bot_messages: List[str]) -> None:
        for message in bot_messages:
            print(f"Bot: {message}")


if __name__ == "__main__":
    Chat(
        CliUserMessageSource(),
        CliOutputWriter(),
        AnswerGenerator(),
        ConversationLogger("logs.txt")
    ).run()

Простой чат для интерфейса командной строки

Эта простая версия позволяет считывать ввод из CLI и выводить ответы, а также логировать беседу в текстовом файле.

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

from dataclasses import dataclass
from random import random
from time import sleep
from typing import List

from .chat import AnswerGenerator, Chat, OutputWriter, UserMessageSource


@dataclass
class MqConfig:
    host: str
    port: int
    username: str
    password: str


class MqUserMessageSource(UserMessageSource):
    def __init__(self, config: MqConfig):
        self.config = config

    def get_user_message(self) -> str:
        return self.poll_messages()

    def poll_messages(self) -> str:
        # Fake method, real implementation would use the MqConfig
        sleep(1)
        return "hi" if random() > 0.2 else "bye"


class MqOutputWriter(OutputWriter):
    def __init__(self, config: MqConfig):
        self.config = config

    def write_bot_messages(self, bot_messages: List[str]) -> None:
        for message in bot_messages:
            self.produce_message(message)

    def produce_message(self, message: str) -> None:
        # Fake method, real implementation would use the MqConfig
        pass


if __name__ == "__main__":
    mq_config = MqConfig(
        "localhost",
        1234,
        "mq_user",
        "my_password",
    )
    Chat(
        MqUserMessageSource(mq_config),
        MqOutputWriter(mq_config),
        AnswerGenerator(),
    ).run()

Сымитированный чат, подключенный через очередь сообщений

В этой версии нам понадобится экземпляр MqConfig, который бы совместно использовался между MqUserMessageSource и MqOutputWriter. ConversationLogger здесь не используется, его можно было бы добавить как независимый потребитель очереди сообщений.

Код работает, но в нем остались некоторые болевые точки:

• Создание новой конфигурации: подразумевается, что при этом будет создаваться новый экземпляр Chat и все его зависимости. Если это решение не кажется вам работоспособным – помните, это всего лишь пример, вряд ли вы стали бы его придерживаться, если бы вам потребовалось работать более чем со 100 классами зависимостей.

• Добавление/удаление/замена возможности: изменение группы логически связанных зависимостей может быть изнурительным (вам придется удалить 2 класса и добавить еще 3, чтобы у вас был чатбот, подключенный к очереди сообщений).

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

• Здесь слишком много шаблонного кода, в нем легко допустить ошибку и ценность его невелика.

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

Как это работает?

Вот несколько примеров, демонстрирующих, как при помощи библиотеки opyoid упростить ваше приложение.

from opyoid import ClassBinding, Injector, InstanceBinding

from .chat import AnswerGenerator, Chat, ConversationLogger, OutputWriter, UserMessageSource
from .cli import CliOutputWriter, CliUserMessageSource


if __name__ == "__main__":
    injector = Injector(bindings=[
        ClassBinding(Chat),
        ClassBinding(AnswerGenerator),
        InstanceBinding(ConversationLogger, ConversationLogger("file.txt")),
        ClassBinding(UserMessageSource, bound_type=CliUserMessageSource),
        ClassBinding(OutputWriter, bound_type=CliOutputWriter),
    ])
    chat = injector.inject(Chat)
    chat.run()

Версия для интерфейса командной строки, использующая библиотеку opyoid

Как видите, тут создаются связки, конфигурирующие, экземпляры каких классов следует создавать – а все остальное делает opyoid.

• Здесь для Chat требуется экземпляр UserMessageSource, а CliUserMessageSource связан с этим типом, поэтому его экземпляр создается, когда это нужно. То же касается OutputWriter, который связан с версией CLI. Когда вы хотите привязать класс к нему же самому, например, Chat или AnswerGenerator, то вам не приходится объявлять их дважды.

• ConversationLogger связан с экземпляром самого себя, он будет использоваться напрямую, когда это станет необходимо.

• Все связки даются Injector, который затем может использовать их для создания экземпляров новых объектов.

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

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

Как же теперь выглядит версия с очередью сообщений?

from opyoid import ClassBinding, Injector, InstanceBinding

from .chat import AnswerGenerator, Chat, OutputWriter, UserMessageSource
from .mq import MqConfig, MqOutputWriter, MqUserMessageSource

if __name__ == "__main__":
    injector = Injector(bindings=[
        ClassBinding(Chat),
        ClassBinding(AnswerGenerator),
        ClassBinding(UserMessageSource, bound_type=MqUserMessageSource),
        ClassBinding(OutputWriter, bound_type=MqOutputWriter),
        InstanceBinding(MqConfig, bound_instance=MqConfig(
            "localhost",
            1234,
            "mq_user",
            "my_password",
        )),
    ])
    chat = injector.inject(Chat)
    chat.run()

Версия с очередью сообщений, использующая opyoid

Обратите внимание, как мы воспользовались InstanceBinding для MqConfig. Это полезно, когда требуется внедрить конфигурационные классы данных, содержащие типы-примитивы – например, строки, целые числа или булевы значения. Этот MqConfig автоматически связан со всеми подключениями очереди сообщений, поэтому при добавлении нового не потребовалось бы никакой дополнительной конфигурации кроме самого класса. Также здесь видно, что при удалении ConversationLogger привязка не представляет проблем, поскольку имеет значение по умолчанию в конструкторе Chat. Нужны только параметры без значений по умолчанию.

Дальнейшие улучшения

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

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

from opyoid import ClassBinding, Injector, InstanceBinding, Provider, ProviderBinding
from typing import Optional

from .chat import AnswerGenerator, Chat, ConversationLogger, OutputWriter, UserMessageSource
from .cli import CliOutputWriter, CliUserMessageSource


class ChatProvider(Provider[Chat]):
    def __init__(self,
                 user_message_source: UserMessageSource,
                 output_writer: OutputWriter,
                 answer_generator: AnswerGenerator,
                 conversation_logger: Optional[ConversationLogger] = None):
        self.user_message_source = user_message_source
        self.output_writer = output_writer
        self.answer_generator = answer_generator
        self.conversation_logger = conversation_logger

    def get(self) -> Chat:
        return Chat(
            self.user_message_source,
            self.output_writer,
            self.answer_generator,
            self.conversation_logger,
        )

if __name__ == "__main__":
    injector = Injector(bindings=[
        ProviderBinding(Chat, bound_provider=ChatProvider),
        ClassBinding(AnswerGenerator),
        InstanceBinding(ConversationLogger, ConversationLogger("file.txt")),
        ClassBinding(UserMessageSource, bound_type=CliUserMessageSource),
        ClassBinding(OutputWriter, bound_type=CliOutputWriter),
    ])
    chat = injector.inject(Chat)
    chat.run()

Провайдер для класса Chat

Здесь ChatProvider будет использоваться для создания каждого необходимого экземпляра Chat, даже если в его конструкторе не будет подсказок типов. Обратите внимание на ProviderBinding в инициализации Injector.

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

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

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

from opyoid import Module

from .chat import AnswerGenerator, Chat, ConversationLogger, OutputWriter, UserMessageSource
from .cli import CliOutputWriter, CliUserMessageSource
from .mq import MqOutputWriter, MqUserMessageSource


class ChatModule(Module):
    def configure(self) -> None:
        self.bind(Chat)
        self.bind(AnswerGenerator)


class CliModule(Module):
    def configure(self) -> None:
        self.bind(ConversationLogger, to_instance=ConversationLogger("file.txt"))
        self.bind(UserMessageSource, to_class=CliUserMessageSource)
        self.bind(OutputWriter, to_class=CliOutputWriter)


class MqModule(Module):
    def configure(self) -> None:
        self.bind(UserMessageSource, to_class=MqUserMessageSource)
        self.bind(OutputWriter, to_class=MqOutputWriter)

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

from opyoid import Injector, Module

from .chat import Chat, ConversationLogger
from .modules import ChatModule, CliModule


class CliChatModule(Module):
    def configure(self) -> None:
        self.install(ChatModule())
        self.install(CliModule())
        self.bind(ConversationLogger, to_instance=ConversationLogger("file.txt"))


if __name__ == "__main__":
    injector = Injector(modules=[CliChatModule()])
    chat = injector.inject(Chat)
    chat.run()

Конфигурация значительно упростилась

В модуле вы можете объявить столько связок, сколько хотите, а также установить другие модули по мере необходимости. Так, здесь мы установили ChatModule в CliChatModule.

Обратите внимание: при этом вы по-прежнему можете использовать связки поверх модуля в вашем инъекторе.

from opyoid import Injector, InstanceBinding, Module

from .chat import Chat
from .modules import ChatModule, MqModule
from .mq import MqConfig


class MqChatModule(Module):
    def configure(self) -> None:
        self.install(ChatModule())
        self.install(MqModule())

if __name__ == "__main__":
    injector = Injector(
        modules=[MqChatModule()],
        bindings=[InstanceBinding(MqConfig, bound_instance=MqConfig(
            "localhost",
            1234,
            "mq_user",
            "my_password",
        ))]
    )
    chat = injector.inject(Chat)
    chat.run()

Версия очереди сообщений с модулями

А вы могли бы сделать лучше? Мы уже говорили о сокращении шаблонного кода, но мне все равно придется писать все эти связки?

Довольно слов:

from opyoid import Injector, InjectorOptions, InstanceBinding

from .chat import Chat, ConversationLogger
from .modules import CliModule

if __name__ == "__main__":
    injector = Injector(
        modules=[CliModule()],
        bindings=[InstanceBinding(ConversationLogger, ConversationLogger("file.txt"))],
        options=InjectorOptions(auto_bindings=True))
    chat = injector.inject(Chat)
    chat.run()

При использовании опции auto_bindings классы автоматически привязываются сами к себе, поэтому вам всего лишь потребуется написать другие связки. Конечно же, вы можете и дальше пользоваться модулями и связками, а такие автосвязки будут создаваться только в качестве последнего варианта. В данном примере мы удалили ChatModule, поскольку он содержал только ClassBindings.

В сравнении с кодом, который был у нас в самом начале, теперь имеем:

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

• Никакого дублирования между конфигурациями

• Количество шаблонного кода сведено до абсолютного минимума

Почему бы не переиспользовать имеющуюся библиотеку?

Мы обнаружили, что в экосистеме Python отсутствуют серьезные кандидаты для этого. Лучшей из библиотек, которую мы протестировали, была pinject, но мы хотели воспользоваться преимуществами типизации при внедрении наших классов. Другая альтернатива - python-dependency-injector, но для ее начальной настройки требуется достаточно много кода.

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

Вот основные цели, которые ставились при создании этой библиотеки:

• Автоматически предоставлять зависимости для каждого класса

• Использовать типизацию для их разрешения, поскольку они становятся нормой в Python

• Иметь возможность внедрять сторонние классы

• Иметь возможность работать без обязательных декораторов во всех классах

Мы также добавили некоторые другие продвинутые возможности, которые нужны в данном случае:

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

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

• Аннотации, чтобы можно было иметь множество связок для одного и того же типа

• Многое другое…

Заключение

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