/usr/share/pyshared/envisage/service_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 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 | """ The service registry. """
# Standard library imports.
import logging
# Enthought library imports.
from traits.api import Dict, Event, HasTraits, Int, Undefined, implements, \
Interface
# Local imports.
from i_service_registry import IServiceRegistry
from import_manager import ImportManager
# Logging.
logger = logging.getLogger(__name__)
class NoSuchServiceError(Exception):
""" Raised when a required service is not found. """
class ServiceRegistry(HasTraits):
""" The service registry. """
implements(IServiceRegistry)
#### IServiceRegistry interface ##########################################
# An event that is fired when a service is registered.
registered = Event
# An event that is fired when a service is unregistered.
unregistered = Event
#### Private interface ###################################################
# The services in the registry.
#
# { service_id : (protocol_name, obj, properties) }
#
# where:
#
# 'protocol_name' is the (possible dotted) name of the interface, type or
# class that the object is registered against.
#
# 'obj' is the object that is registered (any old, Python object!).
#
# 'properties' is the arbitrary dictionary of properties that were
# registered with the object.
_services = Dict
# The next service Id (service Ids are never persisted between process
# invocations so this is simply an ever increasing integer!).
_service_id = Int
###########################################################################
# 'IServiceRegistry' interface.
###########################################################################
def get_required_service(self, protocol, query='', minimize='',maximize=''):
""" Return the service that matches the specified query.
Raise a 'NoSuchServiceError' exception if no such service exists.
"""
service = self.get_service(protocol, query, minimize, maximize)
if service is None:
raise NoSuchServiceError(protocol)
return service
def get_service(self, protocol, query='', minimize='', maximize=''):
""" Return at most one service that matches the specified query. """
services = self.get_services(protocol, query, minimize, maximize)
if len(services) > 0:
service = services[0]
else:
service = None
return service
def get_service_from_id(self, service_id):
""" Return the service with the specified id. """
try:
protocol, obj, properties = self._services[service_id]
except KeyError:
raise ValueError('no service with id <%d>' % service_id)
return obj
def get_services(self, protocol, query='', minimize='', maximize=''):
""" Return all services that match the specified query. """
services = []
for service_id, (name, obj, properties) in self._services.items():
if self._get_protocol_name(protocol) == name:
# If the protocol is a string then we need to import it!
if isinstance(protocol, basestring):
actual_protocol = ImportManager().import_symbol(protocol)
# Otherwise, it is an actual protocol, so just use it!
else:
actual_protocol = protocol
# If the registered service is actually a factory then use it
# to create the actual object.
obj = self._resolve_factory(
actual_protocol, name, obj, properties, service_id
)
# If a query was specified then only add the service if it
# matches it!
if len(query) == 0 or self._eval_query(obj, properties, query):
services.append(obj)
# Are we minimizing or maximising anything? If so then sort the list
# of services by the specified attribute/property.
if minimize != '':
services.sort(None, lambda x: getattr(x, minimize))
elif maximize != '':
services.sort(None, lambda x: getattr(x, maximize), reverse=True)
return services
def get_service_properties(self, service_id):
""" Return the dictionary of properties associated with a service. """
try:
protocol, obj, properties = self._services[service_id]
properties = properties.copy()
except KeyError:
raise ValueError('no service with id <%d>' % service_id)
return properties
def register_service(self, protocol, obj, properties=None):
""" Register a service. """
protocol_name = self._get_protocol_name(protocol)
# Make sure each service gets its own properties dictionary.
if properties is None:
properties = {}
service_id = self._next_service_id()
self._services[service_id] = (protocol_name, obj, properties)
self.registered = service_id
logger.debug('service <%d> registered %s', service_id, protocol_name)
return service_id
def set_service_properties(self, service_id, properties):
""" Set the dictionary of properties associated with a service. """
try:
protocol, obj, old_properties = self._services[service_id]
self._services[service_id] = protocol, obj, properties.copy()
except KeyError:
raise ValueError('no service with id <%d>' % service_id)
return
def unregister_service(self, service_id):
""" Unregister a service. """
try:
protocol, obj, properties = self._services.pop(service_id)
self.unregistered = service_id
logger.debug('service <%d> unregistered', service_id)
except KeyError:
raise ValueError('no service with id <%d>' % service_id)
return
###########################################################################
# Private interface.
###########################################################################
def _create_namespace(self, service, properties):
""" Create a namespace in which to evaluate a query. """
namespace = {}
namespace.update(service.__dict__)
namespace.update(properties)
return namespace
def _eval_query(self, service, properties, query):
""" Evaluate a query over a single service.
Return True if the service matches the query, otherwise return False.
"""
namespace = self._create_namespace(service, properties)
try:
result = eval(query, namespace)
except:
result = False
return result
def _get_protocol_name(self, protocol_or_name):
""" Returns the full class name for a protocol. """
if isinstance(protocol_or_name, basestring):
name = protocol_or_name
else:
name = '%s.%s' % (
protocol_or_name.__module__, protocol_or_name.__name__
)
return name
def _is_service_factory(self, protocol, obj):
""" Is the object a factory for services supporting the protocol? """
# fixme: Should we have a formal notion of service factory with an
# appropriate API, or is this good enough? An API might have lifecycle
# methods to both create and destroy the service?!?
return not isinstance(obj, protocol)
def _next_service_id(self):
""" Returns the next service ID. """
self._service_id += 1
return self._service_id
def _resolve_factory(self, protocol, name, obj, properties, service_id):
""" If 'obj' is a factory then use it to create the actual service. """
# Is the registered service actually a service *factory*?
if self._is_service_factory(protocol, obj):
# A service factory is any callable that takes two arguments, the
# first is the protocol, the second is the (possibly empty)
# dictionary of properties that were registered with the service.
#
# If the factory is specified as a symbol path then import it.
if isinstance(obj, basestring):
obj = ImportManager().import_symbol(obj)
obj = obj(**properties)
# The resulting service object replaces the factory in the cache
# (i.e. the factory will not get called again unless it is
# unregistered first).
self._services[service_id] = (name, obj, properties)
return obj
#### EOF ######################################################################
|