/usr/share/pyshared/envisage/extension_registry.py is in python-envisage 4.4.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | """ A base class for extension registry implementation. """
# Standard library imports.
import logging
# Enthought library imports.
from traits.api import Dict, HasTraits, implements
# Local imports.
from extension_point_changed_event import ExtensionPointChangedEvent
from i_extension_registry import IExtensionRegistry
import safeweakref
from unknown_extension_point import UnknownExtensionPoint
# Logging.
logger = logging.getLogger(__name__)
class ExtensionRegistry(HasTraits):
""" A base class for extension registry implementation. """
implements(IExtensionRegistry)
###########################################################################
# Protected 'ExtensionRegistry' interface.
###########################################################################
# A dictionary of extensions, keyed by extension point.
_extensions = Dict
# The extension points that have been added *explicitly*.
_extension_points = Dict
# Extension listeners.
#
# These are called when extensions are added to or removed from an
# extension point.
#
# e.g. Dict(extension_point, [weakref.ref(callable)])
#
# A listener is any Python callable with the following signature:-
#
# def listener(extension_registry, extension_point_changed_event):
# ...
_listeners = Dict
###########################################################################
# 'IExtensionRegistry' interface.
###########################################################################
def add_extension_point_listener(self, listener, extension_point_id=None):
""" Add a listener for extensions being added or removed. """
listeners = self._listeners.setdefault(extension_point_id, [])
listeners.append(safeweakref.ref(listener))
return
def add_extension_point(self, extension_point):
""" Add an extension point. """
self._extension_points[extension_point.id] = extension_point
logger.debug('extension point <%s> added', extension_point.id)
return
def get_extensions(self, extension_point_id):
""" Return the extensions contributed to an extension point. """
return self._get_extensions(extension_point_id)[:]
def get_extension_point(self, extension_point_id):
""" Return the extension point with the specified Id. """
return self._extension_points.get(extension_point_id)
def get_extension_points(self):
""" Return all extension points. """
return self._extension_points.values()
def remove_extension_point_listener(self,listener,extension_point_id=None):
""" Remove a listener for extensions being added or removed. """
listeners = self._listeners.setdefault(extension_point_id, [])
listeners.remove(safeweakref.ref(listener))
return
def remove_extension_point(self, extension_point_id):
""" Remove an extension point. """
self._check_extension_point(extension_point_id)
# Remove the extension point.
del self._extension_points[extension_point_id]
# Remove any extensions to the extension point.
if extension_point_id in self._extensions:
old = self._extensions[extension_point_id]
del self._extensions[extension_point_id]
else:
old = []
refs = self._get_listener_refs(extension_point_id)
self._call_listeners(refs, extension_point_id, [], old, 0)
logger.debug('extension point <%s> removed', extension_point_id)
return
def set_extensions(self, extension_point_id, extensions):
""" Set the extensions contributed to an extension point. """
self._check_extension_point(extension_point_id)
old = self._get_extensions(extension_point_id)
self._extensions[extension_point_id] = extensions
refs = self._get_listener_refs(extension_point_id)
self._call_listeners(refs, extension_point_id, extensions, old, None)
return
###########################################################################
# Protected 'ExtensionRegistry' interface.
###########################################################################
def _call_listeners(self, refs, extension_point_id, added, removed, index):
""" Call listeners that are listening to an extension point. """
event = ExtensionPointChangedEvent(
extension_point_id = extension_point_id,
added = added,
removed = removed,
index = index
)
for ref in refs:
listener = ref()
if listener is not None:
listener(self, event)
return
def _check_extension_point(self, extension_point_id):
""" Check to see if the extension point exists.
Raise an 'UnknownExtensionPoint' if it does not.
"""
if not extension_point_id in self._extension_points:
raise UnknownExtensionPoint(extension_point_id)
return
def _get_extensions(self, extension_point_id):
""" Return the extensions for the given extension point. """
return self._extensions.setdefault(extension_point_id, [])
def _get_listener_refs(self, extension_point_id):
""" Get weak references to all listeners to an extension point.
Returns a list containing the weak references to those listeners that
are listening to this extension point specifically first, followed by
those that are listening to any extension point.
"""
refs = []
refs.extend(self._listeners.get(extension_point_id, []))
refs.extend(self._listeners.get(None, []))
return refs
#### EOF ######################################################################
|