Python Plugins with Topics

Source Code: PyPlugins

Any infrastructure should always have the capability to extend itself. It would be better if the functionality is added by a contributor - who is not part of the core team. And it doesn’t get in the way of core components - development, compilation and deployment.

The concept of plugin has been around there for quite a while - Visual Studio/VS Code all has plugins (aka extensions). Basic idea here is to add new functionalities - by just deploying a new dll, jar or py modules.

In this proposal using python, we have a root folder named Plugins - the infra would enumerate the Plugins directory to add functions. The functionalities register themselves with a Key - let’s call them TOPICS.

Well why Topics? I am borrowing this idea from Messaging Queue infra like ZMQ and Kafka… So, that we could create a Topic to Function mapping - and we would be able to map caller to a MQ subscriber.

Let me walk through each component:

Base class

Base class that each plugin must inherit from; this class exposes couple of items 1) List of Topics 2) Execute Method - which your plugin should implement.

class IPlugin(object):

    def __init__(self):
        self.description = 'UNKNOWN'
        self.topics = []

    def execute(self, topic, argument):
        """The method that we expect all plugins to implement. This is the
        method that our framework will call
        """
        raise NotImplementedError

Example Plugin : Calculate

Calculate Plugin - exposes two functionalities: add and subtract. For the client its exposed as two topics. The execute function takes in topic and the argument. Based on the topic the respective plugin could dispatch it to sub-functions within the plugin.

class CalculatorPlugin(IPlugin):
    def __init__(self):
        self.description = 'Calculator'
        self.topics = ['Add', 'Subtract']

    def execute(self, topic, args):
        if topic == "Add":
            return self.add(args)

        raise Exception (f'Topic: {topic} has no mapping function')

    def add(self, args):
        count  = 0
        for index in range(0, len(args)):
            count += args[index]
        return count

Service Discovery:

The infrastructure would enumerate the plugin base_directory and try find sub class of IPlugin. Create instance of the sub_class and register to the store: MAP<topic, instance>. Infra should be able to directly call the execute API, on the instance.

class ServiceDiscovery(object):
    def __init__(self, plugin_package_dir='plugins'):
        self.plugin_package_base_dir = plugin_package_dir
        self.plugin_topic_instance_map = {}

        self.enumerate_packages()

    def enumerate_packages(self, package):
        """Recursively walk the supplied package to retrieve all plugins
        """
        imported_package = __import__(package, fromlist=['foo'])

        for _, pluginname, ispkg in pkgutil.iter_modules(imported_package.__path__, imported_package.__name__ + '.'):
            if not ispkg:
                plugin_module = __import__(pluginname, fromlist=['foo'])
                clsmembers = inspect.getmembers(plugin_module, inspect.isclass)
                for (_, c) in clsmembers:
                    # Only add classes that are a sub class of Plugin, but NOT Plugin itself
                    if issubclass(c, IPlugin) & (c is not IPlugin):
                        print(f'    Found plugin class: {c.__module__}.{c.__name__}')
                        cls_instance = c()
                        for topic in cls_instance.topics:
                            print(f'        Registering Topics: {topic}')
                            self.plugin_topic_instance_map[topic] = cls_instance

Execution

Now that we have a Map<Topic,instance>. When a call comes in - it would have a topic and the args. Using the Map, we could get the corresponding instance and call by passing in both the topic and args. This is similar to delegate (C#) or function pointers(in C).

class ServiceDiscovery(object):
    def __init__(self, plugin_package_dir='plugins'):
        ...
        self.plugin_topic_instance_map = {}

    def execute(self, topic, argument):
        if topic not in self.plugin_topic_instance_map:
            raise Exception (f'Topic: {topic} is not registered')

        return self.plugin_topic_instance_map[topic].execute(topic, argument)

Unit Test:

Writing unit test is not optional. Well, I am a fan of TDD 😉

def test_cal_plugin_add_func(self):
    # Arrange
    ser_dis = ServiceDiscovery()
    
    # Action
    result = ser_dis.execute("Add", [1, 2])

    # Assert
    self.assertEqual(result, 3)

Deployment:

Adding a new plugin should be as simple as

• Copy and paste a new directory under the Plugin base directory
• Directory should have a class which implements IPlugin

Conclusion:

Building a comprehensive plugin infrastructure is non-trivial; look at Visual Studio - you could override pretty much anything, starting from adding intellisese to a new compiler tool chain. Here, we are just look at a small tip - to get started - on having a python based plugin. Always starting off any infra project with the idea of extension in mind - is good to ensure cleaner responsibility separation.

From the SOLID principle : O -> our software should be Opened for extension but closed for modifications 😍