How to Create a Python Plugin System with Stevedore

| November 30, 2021 | Python, Software Development | No Comments
How to Create a Python Plugin System with Stevedore
Photo by Kieran Wood on Unsplash

One of the questions that I see and hear often is how to extend applications using a Python plugin system. For a test engineer, this is usually related to hardware abstraction. For others, they may want to separate the core functionality from extensions. Through this method, deployment can be simplified where only the required pieces are installed through their individual package(s). Whatever the reason may be, several libraries are available to tackle this challenge. In this article, a simple abstraction layer will be built using the Stevedore library and we will cover the the main concepts of building plugins.

The code in this article is available at https://github.com/chinghwayu/plugin_tutorial.

NOTE: While the code illustrates relays and a hardware abstraction layer, the same concept can be used for services such as a database with a software abstraction layer.

Plugin Architecture

Regardless of the plugin framework used, the architecture behind the implementation of plugins is mostly the same.

  • Define abstract base class
  • Define plugin class inheriting from base class

Abstract Base Class

For plugins of similar type in a Python plugin system, there usually is some common functionality. Plugins based on a protocol or specification typically have common basic functionality. For example, instruments that adhere to IEEE 488.2 standard must respond to queries for its identification.

Although the abstract part is sometimes omitted, this is an important step as it establishes the basic structure of each plugin by defining which methods must be implemented. In Python, this is accomplished with the abc module.

The example shown defines a basic relay instrument driver with three methods – connect, disconnect, and reconnect. In the __init__ method, the connected attribute is established for status as this is common between all plugins.

For the other abstract methods, only docstrings are needed. There is no need to include pass or other code.

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.
        """

Plugin Class

When defining the actual implementation, each of the plugins will inherit from the plugin base class. In the example, two plugins are defined that inherit from RelayBase.

In the __init__ method, a super() call is made to create the connected attribute which the other methods will update.

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

Loading Plugins

At this point, you may think that this is all that’s needed to establish a Python plugin system. For simple plugins, this may be true. However, consider the following:

  • How do we decide which plugin to call at runtime?
  • How can we easily switch between the implementations?
  • How can we package these plugins for distribution?
  • How can we test the interfaces to ensure the correct implementation is being called?

For simple applications or scripts, we may simply call the specific implementation directly and hardcode this in. But consider the case where you may be sharing code with a teammate in another location and they don’t have the same relay. We don’t want to maintain another version of the same code with the alternate implementation hardcoded. What we need is way to easily switch between implementations without altering any code. The only change that should be made is in a configuration file for which implementation to use at runtime.

Plugin Entry Point

There are several ways to discover and load plugins. With Stevedore, this uses entry points to establish keys that can be queried as pointers to the location of specific code. This is the same mechanism found often in packages that enable the launching of Python code as command line scripts once installed into an environment. While this uses a built in console_scripts type of entry point, custom entry point types can be created for the purpose of managing plugins.

For a full discussion on extending applications with plugins, please watch the PyCon 2013 talk on this subject from the author of Stevedore, Doug Hellman. The presentation compares the approach taken with Stevedore compared to other existing approaches at the time.

When building packages for distribution, we can add entry points so that when it’s installed into a Python environment, the environment immediately knows where the plugins are located from the established keys. Adding entry points is simple. In the setup.py for a package, assign a dictionary to the entry_points parameter of setup.

from setuptools import setup

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

The plugin_tutorial key is used as the plugin namespace. The names for each plugin is defined as relay1 and relay2. The location of the the plugin is defined as the module name and class within the module separated by colon, relay:RelayOne and relay:RelayTwo.

For cases when we are using plugins but don’t need to install as package, we can register them into the entry point cache of Stevedore. This is useful for development purposes and when implementing unit tests.

The example below checks a namespace for the specified entry point. If the entry point doesn’t exist, it will be added.

Registers a stevedore plugin dynamically without needing to install as a package

Managing Plugins

With Stevedore, there are several ways to manage plugins in a Python plugin system.

  • Drivers – Single Name, Single Entry Point
  • Hooks – Single Name, Many Entry Points
  • Extensions – Many Names, Many Entry Points

For this article, the driver approach will be discussed since that is the most common use case. Take a look at the Stevedore documentation that discuss Patterns for Loading for the other methods.

For a driver, we need to call the DriverManager. In the parameter list, only namespace and name are required which are directly related to the entry points. Optional parameters are available and the one used in this example is invoke_on_load. While the relay example only establishes a class attribute, for an actual instrument driver, we usually need to perform some kind of initialization. This can be executed at the time when the plugin is loaded.

Calling DriverManager will return a manager object. The actual driver object can be accessed through the driver property. From this property, we can also create abstracted methods to call the driver methods.

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)

The **kwargs parameter is not used but included to show implementation of how to pass parameters to drivers which may have different initialization parameters.

The @property decorator for the driver method is syntactic sugar to provide a shortcut to the driver object. If this wasn’t provided, we would need to call the driver’s disconnect method as:

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

Putting It Together

For a plugin installed into a Python environment, the entry points have established when the package was installed. Through the abstraction interface, we can decide which plugin to load at runtime. Shown is a unit test that calls both plugins and each of its methods.

In order to run this, we will need to first install as a package.

$ 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

To call plugins not installed through the package installation process, we’ll need to first register and then call the plugins. This is useful for writing unit tests that include a dummy plugin. Shown is the same unit test with the plugin registered.

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

Resources

For additional information:

Tags:, , ,

About The Author

Ching-Hwa Yu

I'm a software engineering veteran developing enterprise platforms.

Leave a Reply