|
"""Simple plug-in mechanism to provide replacement for setuptools plugins""" |
|
import logging |
|
log = logging.getLogger(__name__) |
|
|
|
class Plugin( object ): |
|
"""Base class for plugins to be loaded""" |
|
loaded = False |
|
def __init__( self, name, import_path, check = None, **named ): |
|
"""Register the plug-in""" |
|
self.name = name |
|
self.import_path = import_path |
|
self.check = check |
|
self.registry.append( self ) |
|
self.__dict__.update( named ) |
|
def load( self ): |
|
"""Attempt to load and return our entry point""" |
|
try: |
|
return importByName( self.import_path ) |
|
except ImportError as err: |
|
log.warning( |
|
'Unable to import %s: %s', |
|
self.import_path, |
|
err |
|
) |
|
return None |
|
@classmethod |
|
def match( cls, *args ): |
|
"""Match to return the plugin which is appropriate to load""" |
|
@classmethod |
|
def all( cls ): |
|
"""Iterate over all registered plugins""" |
|
return cls.registry[:] |
|
@classmethod |
|
def by_name( cls, name ): |
|
for instance in cls.all(): |
|
if instance.name == name: |
|
return instance |
|
return None |
|
|
|
def importByName( fullName ): |
|
"""Import a class by name""" |
|
name = fullName.split(".") |
|
moduleName = name[:-1] |
|
className = name[-1] |
|
module = __import__( ".".join(moduleName), {}, {}, moduleName) |
|
return getattr( module, className ) |
|
|
|
|
|
class PlatformPlugin( Plugin ): |
|
"""Platform-level plugin registration""" |
|
registry = [] |
|
@classmethod |
|
def match( cls, key ): |
|
"""Determine what platform module to load |
|
|
|
key -- (sys.platform,os.name) key to load |
|
""" |
|
for possible in key: |
|
|
|
for plugin in cls.registry: |
|
if plugin.name == possible: |
|
return plugin |
|
raise KeyError( """No platform plugin registered for %s"""%(key,)) |
|
|
|
class FormatHandler( Plugin ): |
|
"""Data-type storage-format handler""" |
|
registry = [] |
|
@classmethod |
|
def match( cls, value ): |
|
"""Lookup appropriate handler based on value (a type)""" |
|
key = '%s.%s'%( value.__module__, value.__name__ ) |
|
for plugin in cls.registry: |
|
set = getattr( plugin, 'check', ()) |
|
if set and key in set: |
|
return plugin |
|
return None |
|
|
