\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5×\D6?\BDê\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2€ \00 $\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4ê\E9\F2á\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # orm/instrumentation.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Defines SQLAlchemy's system of class instrumentation. This module is usually not directly visible to user applications, but defines a large part of the ORM's interactivity. instrumentation.py deals with registration of end-user classes for state tracking. It interacts closely with state.py and attributes.py which establish per-instance and per-class-attribute instrumentation, respectively. The class instrumentation system can be customized on a per-class or global basis using the :mod:`sqlalchemy.ext.instrumentation` module, which provides the means to build and specify alternate instrumentation forms. .. versionchanged: 0.8 The instrumentation extension system was moved out of the ORM and into the external :mod:`sqlalchemy.ext.instrumentation` package. When that package is imported, it installs itself within sqlalchemy.orm so that its more comprehensive resolution mechanics take effect. """ from . import base from . import collections from . import exc from . import interfaces from . import state from .. import util _memoized_key_collection = util.group_expirable_memoized_property() class ClassManager(dict): """Tracks state information at the class level.""" MANAGER_ATTR = base.DEFAULT_MANAGER_ATTR STATE_ATTR = base.DEFAULT_STATE_ATTR _state_setter = staticmethod(util.attrsetter(STATE_ATTR)) deferred_scalar_loader = None original_init = object.__init__ factory = None def __init__(self, class_): self.class_ = class_ self.info = {} self.new_init = None self.local_attrs = {} self.originals = {} self._bases = [ mgr for mgr in [ manager_of_class(base) for base in self.class_.__bases__ if isinstance(base, type) ] if mgr is not None ] for base_ in self._bases: self.update(base_) self.dispatch._events._new_classmanager_instance(class_, self) # events._InstanceEventsHold.populate(class_, self) for basecls in class_.__mro__: mgr = manager_of_class(basecls) if mgr is not None: self.dispatch._update(mgr.dispatch) self.manage() self._instrument_init() if "__del__" in class_.__dict__: util.warn( "__del__() method on class %s will " "cause unreachable cycles and memory leaks, " "as SQLAlchemy instrumentation often creates " "reference cycles. Please remove this method." % class_ ) def __hash__(self): return id(self) def __eq__(self, other): return other is self @property def is_mapped(self): return "mapper" in self.__dict__ @_memoized_key_collection def _all_key_set(self): return frozenset(self) @_memoized_key_collection def _collection_impl_keys(self): return frozenset( [attr.key for attr in self.values() if attr.impl.collection] ) @_memoized_key_collection def _scalar_loader_impls(self): return frozenset( [ attr.impl for attr in self.values() if attr.impl.accepts_scalar_loader ] ) @util.memoized_property def mapper(self): # raises unless self.mapper has been assigned raise exc.UnmappedClassError(self.class_) def _all_sqla_attributes(self, exclude=None): """return an iterator of all classbound attributes that are implement :class:`.InspectionAttr`. This includes :class:`.QueryableAttribute` as well as extension types such as :class:`.hybrid_property` and :class:`.AssociationProxy`. """ found = {} # constraints: # 1. yield keys in cls.__dict__ order # 2. if a subclass has the same key as a superclass, include that # key as part of the ordering of the superclass, because an # overridden key is usually installed by the mapper which is going # on a different ordering # 3. don't use getattr() as this fires off descriptors for supercls in self.class_.__mro__[0:-1]: inherits = supercls.__mro__[1] for key in supercls.__dict__: found.setdefault(key, supercls) if key in inherits.__dict__: continue val = found[key].__dict__[key] if ( isinstance(val, interfaces.InspectionAttr) and val.is_attribute ): yield key, val def _get_class_attr_mro(self, key, default=None): """return an attribute on the class without tripping it.""" for supercls in self.class_.__mro__: if key in supercls.__dict__: return supercls.__dict__[key] else: return default def _attr_has_impl(self, key): """Return True if the given attribute is fully initialized. i.e. has an impl. """ return key in self and self[key].impl is not None def _subclass_manager(self, cls): """Create a new ClassManager for a subclass of this ClassManager's class. This is called automatically when attributes are instrumented so that the attributes can be propagated to subclasses against their own class-local manager, without the need for mappers etc. to have already pre-configured managers for the full class hierarchy. Mappers can post-configure the auto-generated ClassManager when needed. """ manager = manager_of_class(cls) if manager is None: manager = _instrumentation_factory.create_manager_for_cls(cls) return manager def _instrument_init(self): # TODO: self.class_.__init__ is often the already-instrumented # __init__ from an instrumented superclass. We still need to make # our own wrapper, but it would # be nice to wrap the original __init__ and not our existing wrapper # of such, since this adds method overhead. self.original_init = self.class_.__init__ self.new_init = _generate_init(self.class_, self) self.install_member("__init__", self.new_init) def _uninstrument_init(self): if self.new_init: self.uninstall_member("__init__") self.new_init = None @util.memoized_property def _state_constructor(self): self.dispatch.first_init(self, self.class_) return state.InstanceState def manage(self): """Mark this instance as the manager for its class.""" setattr(self.class_, self.MANAGER_ATTR, self) def dispose(self): """Disassociate this manager from its class.""" delattr(self.class_, self.MANAGER_ATTR) @util.hybridmethod def manager_getter(self): return _default_manager_getter @util.hybridmethod def state_getter(self): """Return a (instance) -> InstanceState callable. "state getter" callables should raise either KeyError or AttributeError if no InstanceState could be found for the instance. """ return _default_state_getter @util.hybridmethod def dict_getter(self): return _default_dict_getter def instrument_attribute(self, key, inst, propagated=False): if propagated: if key in self.local_attrs: return # don't override local attr with inherited attr else: self.local_attrs[key] = inst self.install_descriptor(key, inst) _memoized_key_collection.expire_instance(self) self[key] = inst for cls in self.class_.__subclasses__(): manager = self._subclass_manager(cls) manager.instrument_attribute(key, inst, True) def subclass_managers(self, recursive): for cls in self.class_.__subclasses__(): mgr = manager_of_class(cls) if mgr is not None and mgr is not self: yield mgr if recursive: for m in mgr.subclass_managers(True): yield m def post_configure_attribute(self, key): _instrumentation_factory.dispatch.attribute_instrument( self.class_, key, self[key] ) def uninstrument_attribute(self, key, propagated=False): if key not in self: return if propagated: if key in self.local_attrs: return # don't get rid of local attr else: del self.local_attrs[key] self.uninstall_descriptor(key) _memoized_key_collection.expire_instance(self) del self[key] for cls in self.class_.__subclasses__(): manager = manager_of_class(cls) if manager: manager.uninstrument_attribute(key, True) def unregister(self): """remove all instrumentation established by this ClassManager.""" self._uninstrument_init() self.mapper = self.dispatch = None self.info.clear() for key in list(self): if key in self.local_attrs: self.uninstrument_attribute(key) def install_descriptor(self, key, inst): if key in (self.STATE_ATTR, self.MANAGER_ATTR): raise KeyError( "%r: requested attribute name conflicts with " "instrumentation attribute of the same name." % key ) setattr(self.class_, key, inst) def uninstall_descriptor(self, key): delattr(self.class_, key) def install_member(self, key, implementation): if key in (self.STATE_ATTR, self.MANAGER_ATTR): raise KeyError( "%r: requested attribute name conflicts with " "instrumentation attribute of the same name." % key ) self.originals.setdefault(key, getattr(self.class_, key, None)) setattr(self.class_, key, implementation) def uninstall_member(self, key): original = self.originals.pop(key, None) if original is not None: setattr(self.class_, key, original) def instrument_collection_class(self, key, collection_class): return collections.prepare_instrumentation(collection_class) def initialize_collection(self, key, state, factory): user_data = factory() adapter = collections.CollectionAdapter( self.get_impl(key), state, user_data ) return adapter, user_data def is_instrumented(self, key, search=False): if search: return key in self else: return key in self.local_attrs def get_impl(self, key): return self[key].impl @property def attributes(self): return iter(self.values()) # InstanceState management def new_instance(self, state=None): instance = self.class_.__new__(self.class_) if state is None: state = self._state_constructor(instance, self) self._state_setter(instance, state) return instance def setup_instance(self, instance, state=None): if state is None: state = self._state_constructor(instance, self) self._state_setter(instance, state) def teardown_instance(self, instance): delattr(instance, self.STATE_ATTR) def _serialize(self, state, state_dict): return _SerializeManager(state, state_dict) def _new_state_if_none(self, instance): """Install a default InstanceState if none is present. A private convenience method used by the __init__ decorator. """ if hasattr(instance, self.STATE_ATTR): return False elif self.class_ is not instance.__class__ and self.is_mapped: # this will create a new ClassManager for the # subclass, without a mapper. This is likely a # user error situation but allow the object # to be constructed, so that it is usable # in a non-ORM context at least. return self._subclass_manager( instance.__class__ )._new_state_if_none(instance) else: state = self._state_constructor(instance, self) self._state_setter(instance, state) return state def has_state(self, instance): return hasattr(instance, self.STATE_ATTR) def has_parent(self, state, key, optimistic=False): """TODO""" return self.get_impl(key).hasparent(state, optimistic=optimistic) def __bool__(self): """All ClassManagers are non-zero regardless of attribute state.""" return True __nonzero__ = __bool__ def __repr__(self): return "<%s of %r at %x>" % ( self.__class__.__name__, self.class_, id(self), ) class _SerializeManager(object): """Provide serialization of a :class:`.ClassManager`. The :class:`.InstanceState` uses ``__init__()`` on serialize and ``__call__()`` on deserialize. """ def __init__(self, state, d): self.class_ = state.class_ manager = state.manager manager.dispatch.pickle(state, d) def __call__(self, state, inst, state_dict): state.manager = manager = manager_of_class(self.class_) if manager is None: raise exc.UnmappedInstanceError( inst, "Cannot deserialize object of type %r - " "no mapper() has " "been configured for this class within the current " "Python process!" % self.class_, ) elif manager.is_mapped and not manager.mapper.configured: manager.mapper._configure_all() # setup _sa_instance_state ahead of time so that # unpickle events can access the object normally. # see [ticket:2362] if inst is not None: manager.setup_instance(inst, state) manager.dispatch.unpickle(state, state_dict) class InstrumentationFactory(object): """Factory for new ClassManager instances.""" def create_manager_for_cls(self, class_): assert class_ is not None assert manager_of_class(class_) is None # give a more complicated subclass # a chance to do what it wants here manager, factory = self._locate_extended_factory(class_) if factory is None: factory = ClassManager manager = factory(class_) self._check_conflicts(class_, factory) manager.factory = factory self.dispatch.class_instrument(class_) return manager def _locate_extended_factory(self, class_): """Overridden by a subclass to do an extended lookup.""" return None, None def _check_conflicts(self, class_, factory): """Overridden by a subclass to test for conflicting factories.""" return def unregister(self, class_): manager = manager_of_class(class_) manager.unregister() manager.dispose() self.dispatch.class_uninstrument(class_) if ClassManager.MANAGER_ATTR in class_.__dict__: delattr(class_, ClassManager.MANAGER_ATTR) # this attribute is replaced by sqlalchemy.ext.instrumentation # when imported. _instrumentation_factory = InstrumentationFactory() # these attributes are replaced by sqlalchemy.ext.instrumentation # when a non-standard InstrumentationManager class is first # used to instrument a class. instance_state = _default_state_getter = base.instance_state instance_dict = _default_dict_getter = base.instance_dict manager_of_class = _default_manager_getter = base.manager_of_class def register_class(class_): """Register class instrumentation. Returns the existing or newly created class manager. """ manager = manager_of_class(class_) if manager is None: manager = _instrumentation_factory.create_manager_for_cls(class_) return manager def unregister_class(class_): """Unregister class instrumentation.""" _instrumentation_factory.unregister(class_) def is_instrumented(instance, key): """Return True if the given attribute on the given instance is instrumented by the attributes package. This function may be used regardless of instrumentation applied directly to the class,