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 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.
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 ofdict
, which we will also go over later.register_plugin
initializes the plugin and adds it into theplugins
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 inplugin_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 ofdict
to create a dictionary.__getitem__
re-implements the parent function so that if aKeyError
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 ∞
- Great breakdown on Python metaclasses
- Python emulation container types
- Good insight into how a dictionary works