Как создать систему плагинов Python с помощью Stevedore

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

Код в этой статье доступен по адресу https://github.com/chinghwayu/plugin_tutorial.

ПРИМЕЧАНИЕ. Хотя код иллюстрирует ретрансляторы и уровень аппаратной абстракции, ту же концепцию можно использовать для таких служб, как база данных с уровнем программной абстракции.

Плагин Архитектура

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

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

Абстрактный базовый класс

Для плагинов подобного типа в системе плагинов Python обычно есть некоторые общие функции. Плагины, основанные на протоколе или спецификации, обычно имеют общие базовые функции. Например, приборы, соответствующие стандарту IEEE 488.2, должны отвечать на запросы для его идентификации.

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

Показанный пример определяет базовый драйвер релейного прибора с тремя методами — подключение, отключение и повторное подключение. В методе __init__ для статуса устанавливается атрибут connected, поскольку он является общим для всех плагинов.

Для других абстрактных методов нужны только строки документации. Нет необходимости включать pass или другой код.

from abc import ABCMeta, abstractmethod


class RelayBase(metaclass=ABCMeta):
    """Base class for relay plugins"""

    def __init__(self) -> None:
        """Define base attributes."""
        self.connected = False

    @abstractmethod
    def disconnect(self) -> None:
        """Disconnects relay."""

    @abstractmethod
    def connect(self) -> None:
        """Connects relay."""

    @abstractmethod
    def reconnect(self, seconds: int) -> None:
        """Disconnects for specified time and reconnects.
        Args:
            seconds (int): Amount of time to sleep between disconnect and connect.
        """

Класс плагина

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

В методе __init__ выполняется вызов super() для создания атрибута connected, который будут обновлять другие методы.

class RelayOne(RelayBase):
    def __init__(self):
        super().__init__()

    def disconnect(self):
        self.connected = False
        print("Disconnected One")

    def connect(self):
        self.connected = True
        print("Connected One")

    def reconnect(self, seconds: int = 5):
        self.seconds = seconds
        self.disconnect()
        print(f"One paused for {seconds} seconds...")
        self.connect()


class RelayTwo(RelayBase):
    def __init__(self):
        super().__init__()

    def disconnect(self):
        self.connected = False
        print("Disconnected Two")

    def connect(self):
        self.connected = True
        print("Connected Two")

    def reconnect(self, seconds: int = 5):
        self.seconds = seconds
        self.disconnect()
        print(f"Two paused for {seconds} seconds...")
        self.connect()

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

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

Для простых приложений или сценариев мы можем просто вызвать конкретную реализацию напрямую и жестко закодировать ее. Но рассмотрим случай, когда вы можете делиться кодом с товарищем по команде в другом месте, и у них нет одного и того же ретранслятора. Мы не хотим поддерживать другую версию того же кода с жестко закодированной альтернативной реализацией. Что нам нужно, так это способ легко переключаться между реализациями без изменения кода. Единственное изменение, которое следует внести, — это файл конфигурации, для которого реализация будет использоваться во время выполнения.

Точка входа плагина

Существует несколько способов обнаружения и загрузки плагинов. В Stevedore для этого используются точки входа для установления ключей, которые можно запрашивать как указатели на расположение определенного кода. Это тот же механизм, который часто встречается в пакетах, которые позволяют запускать код Python в виде сценариев командной строки после установки в среду. Хотя здесь используется встроенный тип точки входа console_scripts, для управления плагинами могут быть созданы пользовательские типы точек входа.

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

При сборке пакетов для распространения мы можем добавить точки входа, чтобы при установке в среду Python среда сразу знала, где находятся плагины, по установленным ключам. Добавить точки входа очень просто. В setup.py для пакета назначьте словарь параметру entry_points пакета setup.

from setuptools import setup

setup(
    entry_points={
        "plugin_tutorial": [
            "relay1 = relay:RelayOne",
            "relay2 = relay:RelayTwo",
        ],
    },
)

Ключ plugin_tutorial используется как пространство имен плагинов. Имена для каждого плагина определяются как relay1 и relay2. Расположение плагина определяется как имя модуля и класс внутри модуля, разделенные двоеточием, relay:RelayOne и relay:RelayTwo.

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

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

Управление плагинами

В Stevedore есть несколько способов управления плагинами в системе плагинов Python.

• Драйверы — единое имя, единая точка входа
• Хуки — одно имя, много точек входа
• Расширения — много имен, много точек входа

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

Для драйвера нам нужно вызвать функцию DriverManager. В списке параметров требуются только namespace и name, которые непосредственно связаны с точками входа. Доступны необязательные параметры, и в этом примере используется invoke_on_load. В то время как в примере с реле устанавливается только атрибут класса, для реального драйвера прибора обычно требуется выполнить какую-то инициализацию. Это может быть выполнено во время загрузки плагина.

Вызов DriverManager вернет объект-менеджер. Доступ к фактическому объекту драйвера можно получить через свойство driver. Из этого свойства мы также можем создавать абстрактные методы для вызова методов драйвера.

from stevedore import driver


class Relay:
    def __init__(self, name="", **kwargs) -> None:
        self._relay_mgr = driver.DriverManager(
            namespace="plugin_tutorial",
            name=name,
            invoke_on_load=True,
            invoke_kwds=kwargs,
        )

    @property
    def driver(self):
        return self._relay_mgr.driver

    def disconnect(self) -> None:
        self.driver.disconnect()

    def connect(self) -> None:
        self.driver.connect()

    def reconnect(self, seconds: int = 5) -> None:
        self.driver.reconnect(seconds)

Параметр **kwargs не используется, но включен для демонстрации реализации того, как передавать параметры драйверам, которые могут иметь другие параметры инициализации.

Декоратор @property для метода driver — это синтаксический сахар, обеспечивающий ярлык для объекта драйвера. Если бы это не было предусмотрено, нам пришлось бы вызывать метод disconnect драйвера как:

r = Relay(name="relay1")
r._relay_mgr.driver.disconnect()

Собираем вместе

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

Чтобы запустить это, нам нужно сначала установить его как пакет.

$ pip install -e /path/to/plugin_tutorial
...
Installing collected packages: plugin-tutorial
  Running setup.py develop for plugin-tutorial
Successfully installed plugin-tutorial
$ pip list | grep plugin_tutorial
plugin-tutorial 1.0.0    /path/to/plugin_tutorial

Код теста:

from relay import Relay


def test_installed_plugin():
    r1 = Relay(name="relay1")
    assert isinstance(r1, Relay)
    assert r1.driver.connected == False
    r1.disconnect()
    assert r1.driver.connected == False
    r1.connect()
    assert r1.driver.connected == True
    r1.reconnect(7)
    assert r1.driver.seconds == 7

    r2 = Relay(name="relay2")
    assert isinstance(r2, Relay)
    assert r2.driver.connected == False
    r2.disconnect()
    assert r2.driver.connected == False
    r2.connect()
    assert r2.driver.connected == True
    r2.reconnect(9)
    assert r2.driver.seconds == 9

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

from relay import Relay
from register_plugin import register_plugin


def test_register_plugin():
    namespace = "plugin_tutorial"
    register_plugin(
        name="relay1",
        namespace=namespace,
        entry_point="relay:RelayOne",
    )
    register_plugin(
        name="relay2",
        namespace=namespace,
        entry_point="relay:RelayTwo",
    )
    r1 = Relay(name="relay1")
    assert isinstance(r1, Relay)
    assert r1.driver.connected == False
    r1.disconnect()
    assert r1.driver.connected == False
    r1.connect()
    assert r1.driver.connected == True
    r1.reconnect(7)
    assert r1.driver.seconds == 7

    r2 = Relay(name="relay2")
    assert isinstance(r2, Relay)
    assert r2.driver.connected == False
    r2.disconnect()
    assert r2.driver.connected == False
    r2.connect()
    assert r2.driver.connected == True
    r2.reconnect(9)
    assert r2.driver.seconds == 9

Ресурсы

Для получения дополнительной информации:

• Код туториала — https://github.com/chinghwayu/plugin_tutorial
• руководство пользователя стивидора — https://docs.openstack.org/stevedore/latest/user/index.html
• Абстрактные базовые классы — https://docs.python.org/3/library/abc.html
• Точки входа — https://docs.python.org/3/library/importlib.metadata.html#entry-points
• Презентация PyCon 2013 по стивидорам — https://youtu.be/7K72DPDOhWo
• Yapsy (еще одна система плагинов) — http://yapsy.sourceforge.net/
• Pluggy (системный тест плагинов Pytest) — https://pluggy.readthedocs.io/en/stable/

Первоначально опубликовано на https://chinghwayu.com 30 ноября 2021 г.