This tutorial will walk you through creating a generic plugin framework in Python using metaclassing. This framework can be easily be dropped into existing code to allow seamless integration of Python plugins. There are essentially 3 parts to this framework:

  1. The mount point – This provides a central location between the plugins and the code that uses them. Properly abstracting this class will allow us to use it in multiple systems with different sets of plugins, without having to change any of the core code. We will also need to provide a way to register (mount) a plugin to this object.
  2. Declaring the mount – This will point plugins towards a specific mount point. We will accomplish this task by using the mount point as a metaclass.

  3. Finding/Using the plugin – This is the object that will contain the plugins. It provides easy access to registered plugins and their exposed methods.

Table of Contents [hide]

Mount Point ∞

First we will define the mount (or extension) point for our framework and expose a registration method for plugins. This will provide a centralized, known location between the plugins and the code that uses them.

import os
import imp

class PlugPyMount(type):
    ''' This class acts as a mount point for our plugins    '''

    # Default path to search for plugins - change with register_plugin_dir
    plugin_path = os.path.join(__file__, 'plugins') 

    def __init__(self, name, bases, attrs):
        ''' Initializing mount, or registering a plugin?    '''
        if not hasattr(self, 'plugins'):
            self.plugins = PlugPyStruct(PlugPyMount)
        else:
            self.register_plugin(self)
    
    def register_plugin(self, plugin):
        ''' Registration logic + append to plugins struct '''
        plugin = plugin() #< Init the plugin
        self.plugins[plugin.__class__.__name__] = plugin
    
    @staticmethod
    def register_plugin_dir(plugin_path):
        ''' This function sets the plugin path to be searched   '''
        if os.path.isdir(plugin_path):
            PlugPyMount.plugin_path = plugin_path
        else:
            raise EnvironmentError('%s is not a directory' % plugin_path)
        
    @staticmethod
    def find_plugins():
        ''' Traverse registered plugin directory and import non-loaded modules  '''
        plugin_path = PlugPyMount.plugin_path
        if not os.path.isdir(plugin_path):
            raise EnvironmentError('%s is not a directory' % plugin_path)
        
        for file_ in os.listdir(plugin_path):
            if file_.endswith('.py') and file_ != '__init__.py':
                module = file_[:-3] #< Strip extension
                mod_obj = globals().get(module)
                if mod_obj is None:
                    f, filename, desc = imp.find_module(
                        module, [plugin_path])
                    globals()[module] = mod_obj = imp.load_module(
                        module, f, filename, desc)

The above object subclasses type, which will allow it to be used as a metaclass (we will go over how later).

  • __init__ provides a way of determining whether we are currently working with the class itself, or a plugin that is using it as a metaclass. PlugPyStruct is just a subclass of dict, which we will also go over later.
  • register_plugin initializes the plugin and adds it into the plugins variable under its class name. You can apply whatever other logic you would like to perform on plugins during initialization here.
  • register_plugin_dir updates the path to use when searching for plugins.

  • find_plugins searches for Python modules in plugin_path and imports them into the global namespace if they are not already there.

Declare Mount ∞

Now that we have created the mount point, we need to provide an easy way to use it. Because the mount point subclasses type, we can use it as a metaclass so that our plugins are instances of the mount point. Exactly what is happening internally is beyond the scope of this tutorial; check out the articles linked at the bottom of this page for further information on metaclasses.

class PlugPy(object):
    ''' Default PlugPy implementation, metaclasses PlugPyMount  '''
    __metaclass__ = PlugPyMount

Not much is happening here, but this is one of the more important and mystifying parts of this framework. We will subclass all of our plugins from this object, which will mount them seamlessly to our mount point when the plugin is imported.

Finding Plugins ∞

Now that we have a mount point and a way to mount plugins, we need to provide a method of finding/importing them. Some of this work is done above in the mount class (find_plugins method), but a core component is missing. This component is a subclass of dict, which re-implements the __getitem__() method so that available plugins will be scanned and imported (if not already imported).

class PlugPyStruct(dict):
    '''
        Subclass dict, re-implement __getitem__ to scan for plugins
        if a requested key is missing
    '''
    def __init__(self, cls, *args, **kwargs):
        '''
            Init, set mount to PlugPyMount master instance
            @param  PlugPyMount cls
        '''
        self.mount = cls
        super(PlugPyStruct, self).__init__(*args,**kwargs)

    def __getitem__(self, key, retry=True, default=False):
        ''' Re-implement __getitem__ to scan for plugins if key missing  '''
        try:
            return super(PlugPyStruct, self).__getitem__(key)
        except KeyError:
            if default != False:
                return default
            elif retry:
                self.mount.find_plugins()
                return self.__getitem__(key, False)
            else:
                raise KeyError(
                    'Plugin "%s" not found in plugin_dir "%s"' % (
                        key, self.mount.plugin_path
                    )
                )

The above code should be fairly straightforward:

  • __init__ registers the mount point for use later, then calls the __init__ method of dict to create a dictionary.
  • __getitem__ re-implements the parent function so that if a KeyError is encountered, the plugins directory is searched and the lookup is performed again. If it fails a second time, KeyError is raised per Python’s sequence protocol.

To make things slightly more intuitive, we can reimplement the other dictionary lookup methods to optionally scan for new plugins before allowing iteration over the plugins. This can be useful if you are building an application that has a dynamic plugin set.

    def set_iter_refresh(self, refresh=True):
        '''
            Toggle flag to search for new plugins before iteration
            @param  bool    refresh     Whether to refresh before iteration
        '''
        self.iter_refresh = refresh

    def __iter__(self):
        ''' Reimplement __iter__ to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).__iter__()
    
    def values(self):
        ''' Reimplement values to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).values()
    
    def keys(self):
        ''' Reimplement keys to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).keys()
    
    def items(self):
        ''' Reimplement items to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).items()
    
    def itervalues(self):
        ''' Reimplement itervalues to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).itervalues()
    
    def iterkeys(self):
        ''' Reimplement iterkeys to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).iterkeys()
    
    def iteritems(self):
        ''' Reimplement iteritems to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).iteritems()

set_iter_refresh allows you to set the iter_refresh variable, which will be used to determine whether or not to scan for new plugins before iteration.
All of the other functions check the iter_refresh variable and scan for new plugins if it is set to True. The output of the parent function is then returned.

Making A Plugin ∞

So now that we’ve made a plugin framework, we need to make some plugins. The below code is a basic example of how to do this:

class TestPlugin(PlugPy):

    initialized = False

    def __init__(self):
        print 'Initializing TestPlugin'
        self.initialized = True

    def run(self):
        print 'Running TestPlugin'
        return True

Using The Plugin ∞

We now have a framework and a plugin, but how do we use it? The below code demonstrates just that:

PlugPy.register_plugin('/path/to/plugin/dir')
PlugPy['TestPlugin'].run()

The above code should output:

> Initializing TestPlugin
> Running TestPlugin

Test Suite ∞

Below is a set of tests designed to verify that everything is working correctly with our new framework. These should drop in to your workflow (such as Buildbot) easily:

import os

from unittest import TestCase
from plugpy import *


class PlugPyTest(TestCase):
    
    PLUGIN_DIR = os.path.join(os.path.dirname(__file__), 'test_plugins')

    def setUp(self, ):
        PlugPyMount.register_plugin_dir(self.PLUGIN_DIR)

    def test_non_dir(self, ):
        ''' Test to make sure that an EnvironmentError is raise with an invalid plugin dir '''
        with self.assertRaises(EnvironmentError):
            PlugPyMount.register_plugin_dir(
                os.path.join(self.PLUGIN_DIR, 'not_a_dir')
            )

    def test_plugin(self, ):
        ''' Test if plugin is correctly initialized '''
        plugin_obj = PlugPy.plugins['TestPlugin']
        self.assertIsInstance(plugin_obj, PlugPy) #< Needs to be of type PlugPy, verifies import
        self.assertTrue(plugin_obj.initialized) #< Plugin was initialized
        self.assertTrue(plugin_obj.run()) #< Methods are exposed
        
    def test_iter(self, ):
        ''' Test to verify correct reimplemention of lookups  '''
        plugins = PlugPy.plugins
        
        #   These should all be zero because the plugins haven't been scanned
        self.assertEqual(len([i for i in plugins]), 0) #< __iter__
        self.assertEqual(len(plugins.values()), 0) #< values
        self.assertEqual(len(plugins.keys()), 0) #< keys
        self.assertEqual(len(plugins.items()), 0) #< itervalues
        self.assertEqual(len([i for i in plugins.iteritems()]), 0) #< iteritems
        self.assertEqual(len([i for i in plugins.itervalues()]), 0) #< itervalues
        self.assertEqual(len([i for i in plugins.iterkeys()]), 0) #< iterkeys
        
        #   Set to scan and retry iters, there should now be one plugin
        PlugPy.plugins.set_iter_refresh()
        self.assertTrue(PlugPy.plugins.iter_refresh)
        self.assertEqual(len([i for i in plugins]), 1) #< __iter__
        self.assertEqual(len(plugins.values()), 1) #< values
        self.assertEqual(len(plugins.keys()), 1) #< keys
        self.assertEqual(len(plugins.items()), 1) #< itervalues
        self.assertEqual(len([i for i in plugins.iteritems()]), 1) #< iteritems
        self.assertEqual(len([i for i in plugins.itervalues()]), 1) #< itervalues
        self.assertEqual(len([i for i in plugins.iterkeys()]), 1) #< iterkeys
        
    def test_wrong_plugin(self, ):
        ''' Test to verify that a KeyError is raised for invalid plugin name '''
        with self.assertRaises(KeyError):
            PlugPy.plugins['ThisIsNotAPluginThatExists'] #< Fake

For this test suite to work, you will need to create a directory called test_plugins and add the below script, along with a blank __init__.py file.

import sys
import os

sys.path.append(os.path.join(dirname(__file__), '..'))

from plugpy import PlugPy

class TestPlugin(PlugPy):
    initialized = False
    def __init__(self):
        self.initialized = True
    def run(self):
        return True

Complete Script ∞

File plugpy.py:

##
#    PlugPy
#         
#    This library provides a drop-in plugin framework
#    
#    @author     David Lasley <dave@dlasley.net>
#    @package    plugpy
#    @license    GPLv3 (http://www.gnu.org/licenses/gpl-3.0.html)

import os
import imp

class PlugPyStruct(dict):
    '''
        Subclass dict, reimplement __getitem__ to scan for plugins
        if a requested key is missing
    '''
    def __init__(self, cls, *args, **kwargs):
        '''
            Init, set mount to PlugPyMount master instance
            @param  PlugPyMount cls
        '''
        self.mount = cls
        self.iter_refresh = False
        super(PlugPyStruct, self).__init__(*args,**kwargs)

    def __getitem__(self, key, retry=True):
        ''' Reimplement __getitem__ to scan for plugins if key missing  '''
        try:
            return super(PlugPyStruct, self).__getitem__(key)
        except KeyError:
            if retry:
                self.mount.find_plugins()
                return self.__getitem__(key, False)
            else:
                raise KeyError(
                    'Plugin "%s" not found in plugin_dir "%s"' % (
                        key, self.mount.plugin_path
                    )
                )
    
    def set_iter_refresh(self, refresh=True):
        '''
            Toggle flag to search for new plugins before iteration
            @param  bool    refresh     Whether to refresh before iteration
        '''
        self.iter_refresh = refresh
    
    def __iter__(self):
        ''' Reimplement __iter__ to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).__iter__()
    
    def values(self):
        ''' Reimplement values to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).values()
    
    def keys(self):
        ''' Reimplement keys to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).keys()
    
    def items(self):
        ''' Reimplement items to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).items()
    
    def itervalues(self):
        ''' Reimplement itervalues to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).itervalues()
    
    def iterkeys(self):
        ''' Reimplement iterkeys to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).iterkeys()
    
    def iteritems(self):
        ''' Reimplement iteritems to allow for optional plugin refresh   '''
        if self.iter_refresh:   self.mount.find_plugins()
        return super(PlugPyStruct, self).iteritems()


class PlugPyMount(type):
    ''' This class acts as a mount point for our plugins    '''
    
    # Defaults to ./plugins, change with register_plugin_dir
    plugin_path = os.path.join(__file__, 'plugins')
    
    def __init__(self, name, bases, attrs):
        ''' Initializing mount, or registering a plugin?    '''
        if not hasattr(self, 'plugins'):
            self.plugins = PlugPyStruct(PlugPyMount)
        else:
            self.register_plugin(self)
    
    def register_plugin(self, plugin):
        ''' Registration logic + append to plugins struct '''
        plugin = plugin() #< Init the plugin
        self.plugins[plugin.__class__.__name__] = plugin
    
    @staticmethod
    def register_plugin_dir(plugin_path):
        ''' This function sets the plugin path to be searched   '''
        if os.path.isdir(plugin_path):
            PlugPyMount.plugin_path = plugin_path
        else:
            raise EnvironmentError('%s is not a directory' % plugin_path)
        
    @staticmethod
    def find_plugins():
        ''' Traverse registered plugin directory and import non-loaded modules  '''
        plugin_path = PlugPyMount.plugin_path
        if not os.path.isdir(plugin_path):
            raise EnvironmentError('%s is not a directory' % plugin_path)
        
        for file_ in os.listdir(plugin_path):
            if file_.endswith('.py') and file_ != '__init__.py':
                module = file_[:-3] #< Strip extension
                mod_obj = globals().get(module)
                if mod_obj is None:
                    f, filename, desc = imp.find_module(
                        module, [plugin_path])
                    globals()[module] = mod_obj = imp.load_module(
                        module, f, filename, desc)


class PlugPy(object):
    ''' Default PlugPy implementation, metaclasses PlugPyMount  '''
    __metaclass__ = PlugPyMount


class TestPlugin(PlugPy):
    initialized = False
    def __init__(self):
        print 'Initializing TestPlugin'
        self.initialized = True
    def run(self):
        print 'Running TestPlugin'
        return True


if __name__ == '__main__':
    PlugPy.register_plugin('/path/to/plugin/dir')
    PlugPy['TestPlugin'].run()

File plugpy_test.py:

##
#    PlugPy Tests
#
#    @author     David Lasley <dave@dlasley.net>
#    @package    plugpy.tests
#    @license    GPLv3 (http://www.gnu.org/licenses/gpl-3.0.html)

import os

from unittest import TestCase
from plugpy import *


class PlugPyTest(TestCase):
    
    PLUGIN_DIR = os.path.join(os.path.dirname(__file__), 'test_plugins')

    def setUp(self, ):
        PlugPyMount.register_plugin_dir(self.PLUGIN_DIR)

    def test_non_dir(self, ):
        ''' Test to make sure that an EnvironmentError is raise with an invalid plugin dir '''
        with self.assertRaises(EnvironmentError):
            PlugPyMount.register_plugin_dir(
                os.path.join(self.PLUGIN_DIR, 'not_a_dir')
            )

    def test_plugin(self, ):
        ''' Test if plugin is correctly initialized '''
        plugin_obj = PlugPy.plugins['TestPlugin']
        self.assertIsInstance(plugin_obj, PlugPy) #< Needs to be of type PlugPy, verifies import
        self.assertTrue(plugin_obj.initialized) #< Plugin was initialized
        self.assertTrue(plugin_obj.run()) #< Methods are exposed
        
    def test_iter(self, ):
        ''' Test to verify correct reimplemention of lookups  '''
        plugins = PlugPy.plugins
        
        #   These should all be zero because the plugins haven't been scanned
        self.assertEqual(len([i for i in plugins]), 0) #< __iter__
        self.assertEqual(len(plugins.values()), 0) #< values
        self.assertEqual(len(plugins.keys()), 0) #< keys
        self.assertEqual(len(plugins.items()), 0) #< itervalues
        self.assertEqual(len([i for i in plugins.iteritems()]), 0) #< iteritems
        self.assertEqual(len([i for i in plugins.itervalues()]), 0) #< itervalues
        self.assertEqual(len([i for i in plugins.iterkeys()]), 0) #< iterkeys
        
        #   Set to scan and retry iters, there should now be one plugin
        PlugPy.plugins.set_iter_refresh()
        self.assertTrue(PlugPy.plugins.iter_refresh)
        self.assertEqual(len([i for i in plugins]), 1) #< __iter__
        self.assertEqual(len(plugins.values()), 1) #< values
        self.assertEqual(len(plugins.keys()), 1) #< keys
        self.assertEqual(len(plugins.items()), 1) #< itervalues
        self.assertEqual(len([i for i in plugins.iteritems()]), 1) #< iteritems
        self.assertEqual(len([i for i in plugins.itervalues()]), 1) #< itervalues
        self.assertEqual(len([i for i in plugins.iterkeys()]), 1) #< iterkeys
        
    def test_wrong_plugin(self, ):
        ''' Test to verify that a KeyError is raised for invalid plugin name '''
        with self.assertRaises(KeyError):
            PlugPy.plugins['ThisIsNotAPluginThatExists'] #< Fake

File test_plugins/test_plugin.py:

##
#    PlugPy - Test plugin
#          
#    @author     David Lasley <dave@dlasley.net>
#    @package    plugpy.tests.test_plugins
#    @license    GPLv3 (http://www.gnu.org/licenses/gpl-3.0.html)

import sys
import os

sys.path.append(os.path.join(dirname(__file__), '..'))

from plugpy import PlugPy

class TestPlugin(PlugPy):
    initialized = False
    def __init__(self):
        self.initialized = True
    def run(self):
        return True

Further Reading ∞

4