1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
# -*- coding: UTF-8 -*-
# Copyright 2013 Red Hat, Inc.
# Part of clufter project
# Licensed under GPLv2 (a copy included | http://gnu.org/licenses/gpl-2.0.txt)
"""Easy (at least for usage) plugin framework"""
__author__ = "Jan Pokorný <jpokorny @at@ Red Hat .dot. com>"
import imp
import logging
from os import extsep, walk
from os.path import abspath, dirname, join, splitext
from contextlib import contextmanager
from sys import modules
from .utils import classproperty, hybridproperty
log = logging.getLogger(__name__)
class PluginRegistry(type):
"""Core of simple plugin framework
This ``superclass'' serves two roles:
(1) metaclass for plugins (and its class hierarchy)
- only a tiny wrapper on top of `type`
(2) base class for particular registries
Both roles are combined due to a very tight connection.
Inspired by http://eli.thegreenplace.net/2012/08/07/ (Fundamental...).
"""
_registries = set() # dynamic tracking of specific registries
#
# these are relevant for use case (1)
#
# non-API
def __new__(registry, name, bases, attrs):
if '__metaclass__' not in attrs and (registry.use_local
or not registry.__module__.startswith(attrs['__module__'])):
# alleged end-use plugin
ret = registry.probe(name, bases, attrs)
else:
if registry not in PluginRegistry._registries:
log.debug("Registering `{0}' as registry"
.format(registry.registry))
# specific registry not yet tracked
# (e.g., specific plugin was imported natively)
registry.setup()
PluginRegistry._registries.add(registry)
if registry.namespace not in modules:
# XXX hack to keep going in the test suite
__import__(registry.namespace)
ret = super(PluginRegistry, registry).__new__(registry, name,
bases, attrs)
return ret
#
# these are relevant for both (1) + (2)
#
@classmethod
def probe(registry, name, bases, attrs):
"""Meta-magic to register plugin"""
try:
ret = registry._plugins[name]
log.info("Probe `{0}' plugin under `{1}' registry: already tracked"
.format(name, registry.registry))
except KeyError:
log.debug("Probe `{0}' plugin under `{1}' registry: yet untracked"
.format(name, registry.registry))
ret = super(PluginRegistry, registry).__new__(registry, name,
bases, attrs)
registry._plugins[name] = ret
finally:
if registry._path_context is not None:
registry._path_mapping[registry._path_context].add(name)
return ret
@hybridproperty
def name(this):
"""Nicer access to __name__"""
return this.__name__
@classproperty
def registry(registry):
"""Registry identifier"""
return registry.__name__
@classproperty
def namespace(registry):
"""Absolute namespace possessed by the particular plugin registry
For a plugin, this denotes to which registry/namespace it belongs.
"""
try:
return registry._namespace
except AttributeError:
registry._namespace = '.'.join((__package__, registry.__name__))
return registry._namespace
#
# these are relevant for use case (2)
#
# non-API
@classmethod
@contextmanager
def _path(registry, path):
"""Temporary path context setup enabling safe sideways use"""
assert registry._path_context is None
registry._path_context = path
yield path, registry._path_mapping.setdefault(path, set())
assert registry._path_context is not None
registry._path_context = None
@classmethod
def _context(registry, paths):
"""Iterate through the paths yielding context along
Context is a pair `(path, list_of_per_path_tracked_plugins_so_far)`.
"""
if not isinstance(paths, (list, tuple)):
if paths is None:
return # explictly asked not to use even implicit path
paths = (paths, )
# inject implicit one
implicit = join(dirname(abspath(__file__)), registry.__name__)
paths = (lambda *x: x)(implicit, *paths)
for path in paths:
with registry._path(path) as context:
yield context
# API
use_local = False
@classmethod
def setup(registry, reset=False):
"""Implicit setup upon first registry involvement or external reset"""
attrs = ('_path_context', None), ('_path_mapping', {}), ('_plugins', {})
if reset:
map(lambda (a, d): setattr(registry, a, d), attrs)
else:
map(lambda (a, d): setattr(registry, a, getattr(registry, a, d)),
attrs)
if reset:
PluginRegistry._registries.discard(registry)
@classmethod
def discover(registry, paths):
"""Find relevant plugins available at the specified path(s)
Returns `{plugin_name: plugin_cls}` mapping of plugins found.
"""
ret = {}
for path, path_plugins in registry._context(paths):
# skip if path already discovered (considered final)
if not len(path_plugins):
# visit *.py files within (and under) the path and probe them
for root, dirs, files in walk(path):
for f in files:
name, ext = splitext(f)
if not name.startswith('_') and ext == extsep + 'py':
mfile, mpath, mdesc = imp.find_module(name, [root])
if not mfile:
log.debug("Omitting `{0}' at `{1}'"
.format(name, root))
continue
mname = '.'.join((registry.namespace, name))
try:
imp.load_module(mname, mfile, mpath, mdesc)
finally:
mfile.close()
path_plugins = registry._path_mapping[path]
# filled as a side-effect of meta-magic, see `__new__`
ret.update({n: registry._plugins[n] for n in path_plugins})
if registry.use_local:
# add "built-in" ones
ret.update({n: p for n, p in registry._plugins.iteritems()
if p.__module__ == registry.__module__})
return ret
|
