Creating A Python Plugin Framework With Metaclasses

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:
# **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.
# **Declaring the mount** – This will point plugins towards a specific mount point. We will accomplish this task by using the mount point as a [[http://en.wikipedia.org/wiki/Metaclass|metaclass]].
# **Finding/Using the plugin** – This is the object that will contain the plugins. It provides easy access to registered plugins and their exposed methods.

[[[TOC]]]

=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.
{{{ lang=python
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. {{{ lang=python 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). {{{ lang=python 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 [[http://docs.python.org/2/reference/datamodel.html#emulating-container-types|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. {{{ lang=python 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: {{{ lang=python 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: {{{ lang=python PlugPy.register_plugin('/path/to/plugin/dir') PlugPy['TestPlugin'].run() }}} The above code should output: {{{ lang=bash > 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 [[http://buildbot.net/|Buildbot]]) easily:
{{{ lang=python
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. {{{ lang=python 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`: {{{ lang=python ## # PlugPy # # This library provides a drop-in plugin framework # # @author David Lasley
# @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’)

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_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`: {{{ lang=python ## # PlugPy - Test plugin # # @author David Lasley
# @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=

* [[http://jakevdp.github.io/blog/2012/12/01/a-primer-on-python-metaclasses/|Great breakdown on Python metaclasses]]
* [[http://docs.python.org/2/reference/datamodel.html#emulating-container-types|Python emulation container types]]
* [[http://pybites.blogspot.com/2008/10/pure-python-dictionary-implementation.html|Good insight into how a dictionary works]]
* [[http://en.wikipedia.org/wiki/Metaclass|Wikipedia article on metaclasses]]

Creating A Python Plugin Framework With Metaclasses

Comments

Leave a Reply Cancel reply