Source code for pyramid.registry

import operator
import threading
from zope.interface import implementer
from zope.interface.registry import Components

from pyramid.decorator import reify
from pyramid.interfaces import IIntrospectable, IIntrospector, ISettings
from pyramid.path import CALLER_PACKAGE, caller_package

[docs] class Registry(Components, dict): """A registry object is an :term:`application registry`. It is used by the framework itself to perform mappings of URLs to view callables, as well as servicing other various framework duties. A registry has its own internal API, but this API is rarely used by Pyramid application developers (it's usually only used by developers of the Pyramid framework and Pyramid addons). But it has a number of attributes that may be useful to application developers within application code, such as ``settings``, which is a dictionary containing application deployment settings. For information about the purpose and usage of the application registry, see :ref:`zca_chapter`. The registry may be used both as an :class:`pyramid.interfaces.IDict` and as a Zope component registry. These two ways of storing configuration are independent. Applications will tend to prefer to store information as key-values whereas addons may prefer to use the component registry to avoid naming conflicts and to provide more complex lookup mechanisms. The application registry is usually accessed as ``request.registry`` in application code. By the time a registry is used to handle requests it should be considered frozen and read-only. Any changes to its internal state should be done with caution and concern for thread-safety. """ # for optimization purposes, if no listeners are listening, don't try # to notify them has_listeners = False _settings = None def __init__(self, package_name=CALLER_PACKAGE, *args, **kw): # add a registry-instance-specific lock, which is used when the lookup # cache is mutated self._lock = threading.Lock() # add a view lookup cache self._clear_view_lookup_cache() if package_name is CALLER_PACKAGE: package_name = caller_package().__name__ Components.__init__(self, package_name, *args, **kw) dict.__init__(self) def _clear_view_lookup_cache(self): self._view_lookup_cache = {} def __bool__(self): # defeat bool determination via dict.__len__ return True @reify def package_name(self): return self.__name__ def registerSubscriptionAdapter(self, *arg, **kw): result = Components.registerSubscriptionAdapter(self, *arg, **kw) self.has_listeners = True return result def registerSelfAdapter( self, required=None, provided=None, name='', info='', event=True ): # registerAdapter analogue which always returns the object itself # when required is matched return self.registerAdapter( lambda x: x, required=required, provided=provided, name=name, info=info, event=event, ) def queryAdapterOrSelf(self, object, interface, default=None): # queryAdapter analogue which returns the object if it implements # the interface, otherwise it will return an adaptation to the # interface if not interface.providedBy(object): return self.queryAdapter(object, interface, default=default) return object def registerHandler(self, *arg, **kw): result = Components.registerHandler(self, *arg, **kw) self.has_listeners = True return result
[docs] def notify(self, *events): if self.has_listeners: # iterating over subscribers assures they get executed [_ for _ in self.subscribers(events, None)]
# backwards compatibility for code that wants to look up a settings # object via ``registry.getUtility(ISettings)`` def _get_settings(self): return self._settings def _set_settings(self, settings): self.registerUtility(settings, ISettings) self._settings = settings settings = property(_get_settings, _set_settings)
@implementer(IIntrospector) class Introspector: def __init__(self): self._refs = {} self._categories = {} self._counter = 0 def add(self, intr): category = self._categories.setdefault(intr.category_name, {}) category[intr.discriminator] = intr category[intr.discriminator_hash] = intr intr.order = self._counter self._counter += 1 def get(self, category_name, discriminator, default=None): category = self._categories.setdefault(category_name, {}) intr = category.get(discriminator, default) return intr def get_category(self, category_name, default=None, sort_key=None): if sort_key is None: sort_key = operator.attrgetter('order') category = self._categories.get(category_name) if category is None: return default values = category.values() values = sorted(set(values), key=sort_key) return [ {'introspectable': intr, 'related': self.related(intr)} for intr in values ] def categorized(self, sort_key=None): L = [] for category_name in self.categories(): L.append( ( category_name, self.get_category(category_name, sort_key=sort_key), ) ) return L def categories(self): return sorted(self._categories.keys()) def remove(self, category_name, discriminator): intr = self.get(category_name, discriminator) if intr is None: return L = self._refs.pop(intr, []) for d in L: L2 = self._refs[d] L2.remove(intr) category = self._categories[intr.category_name] del category[intr.discriminator] del category[intr.discriminator_hash] def _get_intrs_by_pairs(self, pairs): introspectables = [] for pair in pairs: category_name, discriminator = pair intr = self._categories.get(category_name, {}).get(discriminator) if intr is None: raise KeyError((category_name, discriminator)) introspectables.append(intr) return introspectables def relate(self, *pairs): introspectables = self._get_intrs_by_pairs(pairs) relatable = ((x, y) for x in introspectables for y in introspectables) for x, y in relatable: L = self._refs.setdefault(x, []) if x is not y and y not in L: L.append(y) def unrelate(self, *pairs): introspectables = self._get_intrs_by_pairs(pairs) relatable = ((x, y) for x in introspectables for y in introspectables) for x, y in relatable: L = self._refs.get(x, []) if y in L: L.remove(y) def related(self, intr): category_name, discriminator = intr.category_name, intr.discriminator intr = self._categories.get(category_name, {}).get(discriminator) if intr is None: raise KeyError((category_name, discriminator)) return self._refs.get(intr, [])
[docs] @implementer(IIntrospectable) class Introspectable(dict): order = 0 # mutated by introspector.add action_info = None # mutated by self.register def __init__(self, category_name, discriminator, title, type_name): self.category_name = category_name self.discriminator = discriminator self.title = title self.type_name = type_name self._relations = [] def relate(self, category_name, discriminator): self._relations.append((True, category_name, discriminator)) def unrelate(self, category_name, discriminator): self._relations.append((False, category_name, discriminator)) def _assert_resolved(self): assert undefer(self.discriminator) is self.discriminator @property def discriminator_hash(self): self._assert_resolved() return hash(self.discriminator) def __hash__(self): self._assert_resolved() return hash((self.category_name,) + (self.discriminator,)) def __repr__(self): self._assert_resolved() return '<%s category %r, discriminator %r>' % ( self.__class__.__name__, self.category_name, self.discriminator, ) def __bool__(self): return True def register(self, introspector, action_info): self.discriminator = undefer(self.discriminator) self.action_info = action_info introspector.add(self) for relate, category_name, discriminator in self._relations: discriminator = undefer(discriminator) if relate: method = introspector.relate else: method = introspector.unrelate method( (self.category_name, self.discriminator), (category_name, discriminator), )
[docs] class Deferred: """Can be used by a third-party configuration extender to wrap a :term:`discriminator` during configuration if an immediately hashable discriminator cannot be computed because it relies on unresolved values. The function should accept no arguments and should return a hashable discriminator.""" def __init__(self, func): self.func = func @reify def value(self): result = self.func() del self.func return result def resolve(self): return self.value
[docs] def undefer(v): """Function which accepts an object and returns it unless it is a :class:`pyramid.registry.Deferred` instance. If it is an instance of that class, its ``resolve`` method is called, and the result of the method is returned.""" if isinstance(v, Deferred): v = v.resolve() return v
[docs] class predvalseq(tuple): """A subtype of tuple used to represent a sequence of predicate values"""
global_registry = Registry('global')