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.
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:
- Tutorial code – https://github.com/chinghwayu/plugin_tutorial
- stevedore user guide – https://docs.openstack.org/stevedore/latest/user/index.html
- Abstract Base Classes – https://docs.python.org/3/library/abc.html
- Entry points – https://docs.python.org/3/library/importlib.metadata.html#entry-points
- PyCon 2013 presentation on stevedore – https://youtu.be/7K72DPDOhWo
- Yapsy (Yet Another Plugin System) – http://yapsy.sourceforge.net/
- Pluggy (Pytest plugin systest) – https://pluggy.readthedocs.io/en/stable/
The post How to Create a Python Plugin System with Stevedore appeared first on Ching-Hwa Yu.
from Planet Python
via read more
No comments:
Post a Comment