a scan of the full contents of the Session
when in use. [ticket:2442]
+ - [moved] The InstrumentationManager interface
+ and the entire related system of alternate
+ class implementation is now moved out
+ to sqlalchemy.ext.instrumentation. This is
+ a seldom used system that adds significant
+ complexity and overhead to the mechanics of
+ class instrumentation. The new architecture
+ allows it to remain unused until
+ InstrumentationManager is actually imported,
+ at which point it is bootstrapped into
+ the core.
+
- [feature] Major rewrite of relationship()
internals now allow join conditions which
include columns pointing to themselves
:doc:`Deprecated Event Interfaces <orm/deprecated>` |
:doc:`ORM Exceptions <orm/exceptions>` |
:doc:`Horizontal Sharding <orm/extensions/horizontal_shard>` |
+ :doc:`Alternate Instrumentation <orm/extensions/instrumentation>`
SQLAlchemy Core
===============
.. autoclass:: AttributeExtension
:members:
-## In 0.7, move "interfaces" to here.
\ No newline at end of file
.. autoclass:: sqlalchemy.orm.events.InstrumentationEvents
:members:
-Alternate Class Instrumentation
--------------------------------
-
-.. autoclass:: sqlalchemy.orm.interfaces.InstrumentationManager
- :members:
- :undoc-members:
-
-
orderinglist
horizontal_shard
hybrid
+ instrumentation
--- /dev/null
+.. _instrumentation_toplevel:
+
+Alternate Class Instrumentation
+================================
+
+.. automodule:: sqlalchemy.ext.instrumentation
+
+API Reference
+-------------
+
+.. autodata:: INSTRUMENTATION_MANAGER
+
+.. autoclass:: InstrumentationManager
+ :members:
+ :undoc-members:
+
+.. autodata:: instrumentation_finders
+
+.. autoclass:: ExtendedInstrumentationRegistry
+ :show-inheritance:
+ :members:
+
+
+
-"""this example illustrates how to replace SQLAlchemy's class descriptors with
-a user-defined system.
+"""Illustrates customized class instrumentation, using
+the :mod:`sqlalchemy.ext.instrumentation` extension package.
-This sort of thing is appropriate for integration with frameworks that
-redefine class behaviors in their own way, such that SQLA's default
-instrumentation is not compatible.
+In this example, mapped classes are modified to
+store their state in a dictionary attached to an attribute
+named "_goofy_dict", instead of using __dict__.
+this example illustrates how to replace SQLAlchemy's class
+descriptors with a user-defined system.
-The example illustrates redefinition of instrumentation at the class level as
-well as the collection level, and redefines the storage of the class to store
-state within "instance._goofy_dict" instead of "instance.__dict__". Note that
-the default collection implementations can be used with a custom attribute
-system as well.
"""
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, Text,\
ForeignKey
-from sqlalchemy.orm import mapper, relationship, Session,\
- InstrumentationManager
+from sqlalchemy.orm import mapper, relationship, Session
from sqlalchemy.orm.attributes import set_attribute, get_attribute, \
del_attribute
from sqlalchemy.orm.instrumentation import is_instrumented
-from sqlalchemy.orm.collections import collection_adapter
+from sqlalchemy.ext.instrumentation import InstrumentationManager
class MyClassState(InstrumentationManager):
- def __init__(self, cls):
- self.states = {}
-
- def instrument_attribute(self, class_, key, attr):
- pass
-
- def install_descriptor(self, class_, key, attr):
- pass
-
- def uninstall_descriptor(self, class_, key, attr):
- pass
-
- def instrument_collection_class(self, class_, key, collection_class):
- return MyCollection
-
def get_instance_dict(self, class_, instance):
return instance._goofy_dict
def initialize_instance_dict(self, class_, instance):
instance.__dict__['_goofy_dict'] = {}
- def initialize_collection(self, key, state, factory):
- data = factory()
- return MyCollectionAdapter(key, state, data), data
-
def install_state(self, class_, instance, state):
- self.states[id(instance)] = state
+ instance.__dict__['_goofy_dict']['state'] = state
def state_getter(self, class_):
def find(instance):
- return self.states[id(instance)]
+ return instance.__dict__['_goofy_dict']['state']
return find
class MyClass(object):
else:
del self._goofy_dict[key]
-class MyCollectionAdapter(object):
- """An wholly alternative instrumentation implementation."""
-
- def __init__(self, key, state, collection):
- self.key = key
- self.state = state
- self.collection = collection
- setattr(collection, '_sa_adapter', self)
-
- def unlink(self, data):
- setattr(data, '_sa_adapter', None)
-
- def adapt_like_to_iterable(self, obj):
- return iter(obj)
-
- def append_with_event(self, item, initiator=None):
- self.collection.add(item, emit=initiator)
-
- def append_multiple_without_event(self, items):
- self.collection.members.extend(items)
-
- def append_without_event(self, item):
- self.collection.add(item, emit=False)
-
- def remove_with_event(self, item, initiator=None):
- self.collection.remove(item, emit=initiator)
-
- def remove_without_event(self, item):
- self.collection.remove(item, emit=False)
-
- def clear_with_event(self, initiator=None):
- for item in list(self):
- self.remove_with_event(item, initiator)
- def clear_without_event(self):
- for item in list(self):
- self.remove_without_event(item)
- def __iter__(self):
- return iter(self.collection)
-
- def fire_append_event(self, item, initiator=None):
- if initiator is not False and item is not None:
- self.state.get_impl(self.key).\
- fire_append_event(self.state, self.state.dict, item,
- initiator)
-
- def fire_remove_event(self, item, initiator=None):
- if initiator is not False and item is not None:
- self.state.get_impl(self.key).\
- fire_remove_event(self.state, self.state.dict, item,
- initiator)
-
- def fire_pre_remove_event(self, initiator=None):
- self.state.get_impl(self.key).\
- fire_pre_remove_event(self.state, self.state.dict,
- initiator)
-
-class MyCollection(object):
- def __init__(self):
- self.members = list()
- def add(self, object, emit=None):
- self.members.append(object)
- collection_adapter(self).fire_append_event(object, emit)
- def remove(self, object, emit=None):
- collection_adapter(self).fire_pre_remove_event(object)
- self.members.remove(object)
- collection_adapter(self).fire_remove_event(object, emit)
- def __getitem__(self, index):
- return self.members[index]
- def __iter__(self):
- return iter(self.members)
- def __len__(self):
- return len(self.members)
if __name__ == '__main__':
meta = MetaData(create_engine('sqlite://'))
assert a1.name == 'a1'
assert a1.bs[0].name == 'b1'
- assert isinstance(a1.bs, MyCollection)
sess = Session()
sess.add(a1)
assert a1.name == 'a1'
assert a1.bs[0].name == 'b1'
- assert isinstance(a1.bs, MyCollection)
a1.bs.remove(a1.bs[0])
--- /dev/null
+"""Extensible class instrumentation.
+
+The :mod:`sqlalchemy.ext.instrumentation` package provides for alternate
+systems of class instrumentation within the ORM. Class instrumentation
+refers to how the ORM places attributes on the class which maintain
+data and track changes to that data, as well as event hooks installed
+on the class.
+
+.. note::
+ The extension package is provided for the benefit of integration
+ with other object management packages, which already perform
+ their own instrumentation. It is not intended for general use.
+
+For examples of how the instrumentation extension is used,
+see the example :ref:`examples_instrumentation`.
+
+.. versionchanged:: 0.8
+ The :mod:`sqlalchemy.orm.instrumentation` was split out so
+ that all functionality having to do with non-standard
+ instrumentation was moved out to :mod:`sqlalchemy.ext.instrumentation`.
+ When imported, the module installs itself within
+ :mod:`sqlalchemy.orm.instrumentation` so that it
+ takes effect, including recognition of
+ ``__sa_instrumentation_manager__`` on mapped classes, as
+ well :attr:`.instrumentation_finders`
+ being used to determine class instrumentation resolution.
+
+"""
+from ..orm import instrumentation as orm_instrumentation
+from ..orm.instrumentation import (
+ ClassManager, InstrumentationFactory, _default_state_getter,
+ _default_dict_getter, _default_manager_getter
+)
+from ..orm import attributes, collections
+from .. import util
+from ..orm import exc as orm_exc
+import weakref
+
+INSTRUMENTATION_MANAGER = '__sa_instrumentation_manager__'
+"""Attribute, elects custom instrumentation when present on a mapped class.
+
+Allows a class to specify a slightly or wildly different technique for
+tracking changes made to mapped attributes and collections.
+
+Only one instrumentation implementation is allowed in a given object
+inheritance hierarchy.
+
+The value of this attribute must be a callable and will be passed a class
+object. The callable must return one of:
+
+ - An instance of an InstrumentationManager or subclass
+ - An object implementing all or some of InstrumentationManager (TODO)
+ - A dictionary of callables, implementing all or some of the above (TODO)
+ - An instance of a ClassManager or subclass
+
+This attribute is consulted by SQLAlchemy instrumentation
+resolution, once the :mod:`sqlalchemy.ext.instrumentation` module
+has been imported. If custom finders are installed in the global
+instrumentation_finders list, they may or may not choose to honor this
+attribute.
+
+"""
+
+def find_native_user_instrumentation_hook(cls):
+ """Find user-specified instrumentation management for a class."""
+ return getattr(cls, INSTRUMENTATION_MANAGER, None)
+
+instrumentation_finders = [find_native_user_instrumentation_hook]
+"""An extensible sequence of callables which return instrumentation implementations
+
+When a class is registered, each callable will be passed a class object.
+If None is returned, the
+next finder in the sequence is consulted. Otherwise the return must be an
+instrumentation factory that follows the same guidelines as
+sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER.
+
+By default, the only finder is find_native_user_instrumentation_hook, which
+searches for INSTRUMENTATION_MANAGER. If all finders return None, standard
+ClassManager instrumentation is used.
+
+"""
+
+class ExtendedInstrumentationRegistry(InstrumentationFactory):
+ """Extends :class:`.InstrumentationFactory` with additional
+ bookkeeping, to accommodate multiple types of
+ class managers.
+
+ """
+ _manager_finders = weakref.WeakKeyDictionary()
+ _state_finders = weakref.WeakKeyDictionary()
+ _dict_finders = weakref.WeakKeyDictionary()
+ _extended = False
+
+ def _locate_extended_factory(self, class_):
+ for finder in instrumentation_finders:
+ factory = finder(class_)
+ if factory is not None:
+ manager = self._extended_class_manager(class_, factory)
+ return manager, factory
+ else:
+ return None, None
+
+ def _check_conflicts(self, class_, factory):
+ existing_factories = self._collect_management_factories_for(class_).\
+ difference([factory])
+ if existing_factories:
+ raise TypeError(
+ "multiple instrumentation implementations specified "
+ "in %s inheritance hierarchy: %r" % (
+ class_.__name__, list(existing_factories)))
+
+ def _extended_class_manager(self, class_, factory):
+ manager = factory(class_)
+ if not isinstance(manager, ClassManager):
+ manager = _ClassInstrumentationAdapter(class_, manager)
+
+ if factory != ClassManager and not self._extended:
+ # somebody invoked a custom ClassManager.
+ # reinstall global "getter" functions with the more
+ # expensive ones.
+ self._extended = True
+ _install_instrumented_lookups()
+
+ self._manager_finders[class_] = manager.manager_getter()
+ self._state_finders[class_] = manager.state_getter()
+ self._dict_finders[class_] = manager.dict_getter()
+ return manager
+
+ def _collect_management_factories_for(self, cls):
+ """Return a collection of factories in play or specified for a
+ hierarchy.
+
+ Traverses the entire inheritance graph of a cls and returns a
+ collection of instrumentation factories for those classes. Factories
+ are extracted from active ClassManagers, if available, otherwise
+ instrumentation_finders is consulted.
+
+ """
+ hierarchy = util.class_hierarchy(cls)
+ factories = set()
+ for member in hierarchy:
+ manager = self.manager_of_class(member)
+ if manager is not None:
+ factories.add(manager.factory)
+ else:
+ for finder in instrumentation_finders:
+ factory = finder(member)
+ if factory is not None:
+ break
+ else:
+ factory = None
+ factories.add(factory)
+ factories.discard(None)
+ return factories
+
+ def unregister(self, class_):
+ if class_ in self._manager_finders:
+ del self._manager_finders[class_]
+ del self._state_finders[class_]
+ del self._dict_finders[class_]
+ super(ExtendedInstrumentationRegistry, self).unregister(class_)
+
+ def manager_of_class(self, cls):
+ if cls is None:
+ return None
+ return self._manager_finders.get(cls, _default_manager_getter)(cls)
+
+ def state_of(self, instance):
+ if instance is None:
+ raise AttributeError("None has no persistent state.")
+ return self._state_finders.get(instance.__class__, _default_state_getter)(instance)
+
+ def dict_of(self, instance):
+ if instance is None:
+ raise AttributeError("None has no persistent state.")
+ return self._dict_finders.get(instance.__class__, _default_dict_getter)(instance)
+
+orm_instrumentation._instrumentation_factory = \
+ _instrumentation_factory = ExtendedInstrumentationRegistry()
+orm_instrumentation.instrumentation_finders = instrumentation_finders
+
+class InstrumentationManager(object):
+ """User-defined class instrumentation extension.
+
+ :class:`.InstrumentationManager` can be subclassed in order
+ to change
+ how class instrumentation proceeds. This class exists for
+ the purposes of integration with other object management
+ frameworks which would like to entirely modify the
+ instrumentation methodology of the ORM, and is not intended
+ for regular usage. For interception of class instrumentation
+ events, see :class:`.InstrumentationEvents`.
+
+ The API for this class should be considered as semi-stable,
+ and may change slightly with new releases.
+
+ .. versionchanged:: 0.8
+ :class:`.InstrumentationManager` was moved from
+ :mod:`sqlalchemy.orm.instrumentation` to
+ :mod:`sqlalchemy.ext.instrumentation`.
+
+ """
+
+ # r4361 added a mandatory (cls) constructor to this interface.
+ # given that, perhaps class_ should be dropped from all of these
+ # signatures.
+
+ def __init__(self, class_):
+ pass
+
+ def manage(self, class_, manager):
+ setattr(class_, '_default_class_manager', manager)
+
+ def dispose(self, class_, manager):
+ delattr(class_, '_default_class_manager')
+
+ def manager_getter(self, class_):
+ def get(cls):
+ return cls._default_class_manager
+ return get
+
+ def instrument_attribute(self, class_, key, inst):
+ pass
+
+ def post_configure_attribute(self, class_, key, inst):
+ pass
+
+ def install_descriptor(self, class_, key, inst):
+ setattr(class_, key, inst)
+
+ def uninstall_descriptor(self, class_, key):
+ delattr(class_, key)
+
+ def install_member(self, class_, key, implementation):
+ setattr(class_, key, implementation)
+
+ def uninstall_member(self, class_, key):
+ delattr(class_, key)
+
+ def instrument_collection_class(self, class_, key, collection_class):
+ return collections.prepare_instrumentation(collection_class)
+
+ def get_instance_dict(self, class_, instance):
+ return instance.__dict__
+
+ def initialize_instance_dict(self, class_, instance):
+ pass
+
+ def install_state(self, class_, instance, state):
+ setattr(instance, '_default_state', state)
+
+ def remove_state(self, class_, instance):
+ delattr(instance, '_default_state')
+
+ def state_getter(self, class_):
+ return lambda instance: getattr(instance, '_default_state')
+
+ def dict_getter(self, class_):
+ return lambda inst: self.get_instance_dict(class_, inst)
+
+class _ClassInstrumentationAdapter(ClassManager):
+ """Adapts a user-defined InstrumentationManager to a ClassManager."""
+
+ def __init__(self, class_, override):
+ self._adapted = override
+ self._get_state = self._adapted.state_getter(class_)
+ self._get_dict = self._adapted.dict_getter(class_)
+
+ ClassManager.__init__(self, class_)
+
+ def manage(self):
+ self._adapted.manage(self.class_, self)
+
+ def dispose(self):
+ self._adapted.dispose(self.class_)
+
+ def manager_getter(self):
+ return self._adapted.manager_getter(self.class_)
+
+ def instrument_attribute(self, key, inst, propagated=False):
+ ClassManager.instrument_attribute(self, key, inst, propagated)
+ if not propagated:
+ self._adapted.instrument_attribute(self.class_, key, inst)
+
+ def post_configure_attribute(self, key):
+ super(_ClassInstrumentationAdapter, self).post_configure_attribute(key)
+ self._adapted.post_configure_attribute(self.class_, key, self[key])
+
+ def install_descriptor(self, key, inst):
+ self._adapted.install_descriptor(self.class_, key, inst)
+
+ def uninstall_descriptor(self, key):
+ self._adapted.uninstall_descriptor(self.class_, key)
+
+ def install_member(self, key, implementation):
+ self._adapted.install_member(self.class_, key, implementation)
+
+ def uninstall_member(self, key):
+ self._adapted.uninstall_member(self.class_, key)
+
+ def instrument_collection_class(self, key, collection_class):
+ return self._adapted.instrument_collection_class(
+ self.class_, key, collection_class)
+
+ def initialize_collection(self, key, state, factory):
+ delegate = getattr(self._adapted, 'initialize_collection', None)
+ if delegate:
+ return delegate(key, state, factory)
+ else:
+ return ClassManager.initialize_collection(self, key,
+ state, factory)
+
+ def new_instance(self, state=None):
+ instance = self.class_.__new__(self.class_)
+ self.setup_instance(instance, state)
+ return instance
+
+ 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 self.has_state(instance):
+ return False
+ else:
+ return self.setup_instance(instance)
+
+ def setup_instance(self, instance, state=None):
+ self._adapted.initialize_instance_dict(self.class_, instance)
+
+ if state is None:
+ state = self._state_constructor(instance, self)
+
+ # the given instance is assumed to have no state
+ self._adapted.install_state(self.class_, instance, state)
+ return state
+
+ def teardown_instance(self, instance):
+ self._adapted.remove_state(self.class_, instance)
+
+ def has_state(self, instance):
+ try:
+ state = self._get_state(instance)
+ except orm_exc.NO_STATE:
+ return False
+ else:
+ return True
+
+ def state_getter(self):
+ return self._get_state
+
+ def dict_getter(self):
+ return self._get_dict
+
+def _install_instrumented_lookups():
+ """Replace global class/object management functions
+ with ExtendedInstrumentationRegistry implementations, which
+ allow multiple types of class managers to be present,
+ at the cost of performance.
+
+ This function is called only by ExtendedInstrumentationRegistry
+ and unit tests specific to this behavior.
+
+ The _reinstall_default_lookups() function can be called
+ after this one to re-establish the default functions.
+
+ """
+ _install_lookups(
+ dict(
+ instance_state = _instrumentation_factory.state_of,
+ instance_dict = _instrumentation_factory.dict_of,
+ manager_of_class = _instrumentation_factory.manager_of_class
+ )
+ )
+
+def _reinstall_default_lookups():
+ """Restore simplified lookups."""
+ _install_lookups(
+ dict(
+ instance_state = _default_state_getter,
+ instance_dict = _default_dict_getter,
+ manager_of_class = _default_manager_getter
+ )
+ )
+
+def _install_lookups(lookups):
+ global instance_state, instance_dict, manager_of_class
+ instance_state = lookups['instance_state']
+ instance_dict = lookups['instance_dict']
+ manager_of_class = lookups['manager_of_class']
+ attributes.instance_state = \
+ orm_instrumentation.instance_state = instance_state
+ attributes.instance_dict = \
+ orm_instrumentation.instance_dict = instance_dict
+ attributes.manager_of_class = \
+ orm_instrumentation.manager_of_class = manager_of_class
from .interfaces import (
EXT_CONTINUE,
EXT_STOP,
- InstrumentationManager,
MapperExtension,
PropComparator,
SessionExtension,
from ..sql import util as sql_util
from .. import util as sa_util
+from . import interfaces
+
+# here, we can establish InstrumentationManager back
+# in sqlalchemy.orm and sqlalchemy.orm.interfaces, which
+# also re-establishes the extended instrumentation system.
+#from ..ext import instrumentation as _ext_instrumentation
+#InstrumentationManager = \
+# interfaces.InstrumentationManager = \
+# _ext_instrumentation.InstrumentationManager
+
__all__ = (
'EXT_CONTINUE',
'EXT_STOP',
- 'InstrumentationManager',
'MapperExtension',
'AttributeExtension',
'PropComparator',
from .. import util, event, inspection
from . import interfaces, collections, events, exc as orm_exc
+from .instrumentation import instance_state, instance_dict, manager_of_class
+
orm_util = util.importlater("sqlalchemy.orm", "util")
PASSIVE_NO_RESULT = util.symbol('PASSIVE_NO_RESULT',
and attributes.py which establish per-instance and per-class-attribute
instrumentation, respectively.
-SQLA's instrumentation system is completely customizable, in which
-case an understanding of the general mechanics of this module is helpful.
-An example of full customization is in /examples/custom_attributes.
+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 exc, collections, events, state, attributes
+from . import exc, collections, events
from operator import attrgetter
from .. import event, util
import weakref
-
-
-INSTRUMENTATION_MANAGER = '__sa_instrumentation_manager__'
-"""Attribute, elects custom instrumentation when present on a mapped class.
-
-Allows a class to specify a slightly or wildly different technique for
-tracking changes made to mapped attributes and collections.
-
-Only one instrumentation implementation is allowed in a given object
-inheritance hierarchy.
-
-The value of this attribute must be a callable and will be passed a class
-object. The callable must return one of:
-
- - An instance of an interfaces.InstrumentationManager or subclass
- - An object implementing all or some of InstrumentationManager (TODO)
- - A dictionary of callables, implementing all or some of the above (TODO)
- - An instance of a ClassManager or subclass
-
-interfaces.InstrumentationManager is public API and will remain stable
-between releases. ClassManager is not public and no guarantees are made
-about stability. Caveat emptor.
-
-This attribute is consulted by the default SQLAlchemy instrumentation
-resolution code. If custom finders are installed in the global
-instrumentation_finders list, they may or may not choose to honor this
-attribute.
-
-"""
-
-instrumentation_finders = []
-"""An extensible sequence of instrumentation implementation finding callables.
-
-Finders callables will be passed a class object. If None is returned, the
-next finder in the sequence is consulted. Otherwise the return must be an
-instrumentation factory that follows the same guidelines as
-INSTRUMENTATION_MANAGER.
-
-By default, the only finder is find_native_user_instrumentation_hook, which
-searches for INSTRUMENTATION_MANAGER. If all finders return None, standard
-ClassManager instrumentation is used.
-
-"""
-
+state = util.importlater("sqlalchemy.orm", "state")
class ClassManager(dict):
"""tracks state information at the class level."""
original_init = object.__init__
+ factory = None
+
def __init__(self, class_):
self.class_ = class_
- self.factory = None # where we came from, for inheritance bookkeeping
self.info = {}
self.new_init = None
self.local_attrs = {}
"""
manager = manager_of_class(cls)
if manager is None:
- manager = _create_manager_for_cls(cls, _source=self)
+ manager = _instrumentation_factory.create_manager_for_cls(cls)
return manager
def _instrument_init(self):
delattr(self.class_, self.MANAGER_ATTR)
+ @util.hybridmethod
def manager_getter(self):
- return attrgetter(self.MANAGER_ATTR)
+ def manager_of_class(cls):
+ return cls.__dict__.get(ClassManager.MANAGER_ATTR, None)
+ return manager_of_class
+
+ @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 attrgetter(self.STATE_ATTR)
+
+ @util.hybridmethod
+ def dict_getter(self):
+ return attrgetter('__dict__')
def instrument_attribute(self, key, inst, propagated=False):
if propagated:
yield m
def post_configure_attribute(self, key):
- instrumentation_registry.dispatch.\
+ _instrumentation_factory.dispatch.\
attribute_instrument(self.class_, key, self[key])
def uninstrument_attribute(self, key, propagated=False):
setattr(instance, self.STATE_ATTR, state)
return state
- 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 attrgetter(self.STATE_ATTR)
-
- def dict_getter(self):
- return attrgetter('__dict__')
-
def has_state(self, instance):
return hasattr(instance, self.STATE_ATTR)
return '<%s of %r at %x>' % (
self.__class__.__name__, self.class_, id(self))
-class _ClassInstrumentationAdapter(ClassManager):
- """Adapts a user-defined InstrumentationManager to a ClassManager."""
-
- def __init__(self, class_, override, **kw):
- self._adapted = override
- self._get_state = self._adapted.state_getter(class_)
- self._get_dict = self._adapted.dict_getter(class_)
-
- ClassManager.__init__(self, class_, **kw)
-
- def manage(self):
- self._adapted.manage(self.class_, self)
-
- def dispose(self):
- self._adapted.dispose(self.class_)
-
- def manager_getter(self):
- return self._adapted.manager_getter(self.class_)
-
- def instrument_attribute(self, key, inst, propagated=False):
- ClassManager.instrument_attribute(self, key, inst, propagated)
- if not propagated:
- self._adapted.instrument_attribute(self.class_, key, inst)
-
- def post_configure_attribute(self, key):
- super(_ClassInstrumentationAdapter, self).post_configure_attribute(key)
- self._adapted.post_configure_attribute(self.class_, key, self[key])
-
- def install_descriptor(self, key, inst):
- self._adapted.install_descriptor(self.class_, key, inst)
-
- def uninstall_descriptor(self, key):
- self._adapted.uninstall_descriptor(self.class_, key)
+class InstrumentationFactory(object):
+ """Factory for new ClassManager instances."""
- def install_member(self, key, implementation):
- self._adapted.install_member(self.class_, key, implementation)
-
- def uninstall_member(self, key):
- self._adapted.uninstall_member(self.class_, key)
-
- def instrument_collection_class(self, key, collection_class):
- return self._adapted.instrument_collection_class(
- self.class_, key, collection_class)
-
- def initialize_collection(self, key, state, factory):
- delegate = getattr(self._adapted, 'initialize_collection', None)
- if delegate:
- return delegate(key, state, factory)
- else:
- return ClassManager.initialize_collection(self, key,
- state, factory)
+ dispatch = event.dispatcher(events.InstrumentationEvents)
- def new_instance(self, state=None):
- instance = self.class_.__new__(self.class_)
- self.setup_instance(instance, state)
- return instance
+ def create_manager_for_cls(self, class_):
+ assert class_ is not None
+ assert manager_of_class(class_) is None
- def _new_state_if_none(self, instance):
- """Install a default InstanceState if none is present.
+ # give a more complicated subclass
+ # a chance to do what it wants here
+ manager, factory = self._locate_extended_factory(class_)
- A private convenience method used by the __init__ decorator.
- """
- if self.has_state(instance):
- return False
- else:
- return self.setup_instance(instance)
+ if factory is None:
+ factory = ClassManager
+ manager = factory(class_)
- def setup_instance(self, instance, state=None):
- self._adapted.initialize_instance_dict(self.class_, instance)
+ self._check_conflicts(class_, factory)
- if state is None:
- state = self._state_constructor(instance, self)
+ manager.factory = factory
- # the given instance is assumed to have no state
- self._adapted.install_state(self.class_, instance, state)
- return state
+ self.dispatch.class_instrument(class_)
+ return manager
- def teardown_instance(self, instance):
- self._adapted.remove_state(self.class_, instance)
+ def _locate_extended_factory(self, class_):
+ """Overridden by a subclass to do an extended lookup."""
+ return None, None
- def has_state(self, instance):
- try:
- state = self._get_state(instance)
- except exc.NO_STATE:
- return False
- else:
- return True
+ def _check_conflicts(self, class_, factory):
+ """Overridden by a subclass to test for conflicting factories."""
+ return
- def state_getter(self):
- return self._get_state
+ 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)
- def dict_getter(self):
- return self._get_dict
+# this attribute is replaced by sqlalchemy.ext.instrumentation
+# when importred.
+_instrumentation_factory = InstrumentationFactory()
-def register_class(class_, **kw):
+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 = _create_manager_for_cls(class_, **kw)
+ manager = _instrumentation_factory.create_manager_for_cls(class_)
return manager
def unregister_class(class_):
"""Unregister class instrumentation."""
- instrumentation_registry.unregister(class_)
+ _instrumentation_factory.unregister(class_)
def is_instrumented(instance, key):
return manager_of_class(instance.__class__).\
is_instrumented(key, search=True)
-class InstrumentationRegistry(object):
- """Private instrumentation registration singleton.
-
- All classes are routed through this registry
- when first instrumented, however the InstrumentationRegistry
- is not actually needed unless custom ClassManagers are in use.
-
- """
-
- _manager_finders = weakref.WeakKeyDictionary()
- _state_finders = util.WeakIdentityMapping()
- _dict_finders = util.WeakIdentityMapping()
- _extended = False
-
- dispatch = event.dispatcher(events.InstrumentationEvents)
-
- def create_manager_for_cls(self, class_, **kw):
- assert class_ is not None
- assert manager_of_class(class_) is None
-
- for finder in instrumentation_finders:
- factory = finder(class_)
- if factory is not None:
- break
- else:
- factory = ClassManager
-
- existing_factories = self._collect_management_factories_for(class_).\
- difference([factory])
- if existing_factories:
- raise TypeError(
- "multiple instrumentation implementations specified "
- "in %s inheritance hierarchy: %r" % (
- class_.__name__, list(existing_factories)))
-
- manager = factory(class_)
- if not isinstance(manager, ClassManager):
- manager = _ClassInstrumentationAdapter(class_, manager)
-
- if factory != ClassManager and not self._extended:
- # somebody invoked a custom ClassManager.
- # reinstall global "getter" functions with the more
- # expensive ones.
- self._extended = True
- _install_lookup_strategy(self)
-
- manager.factory = factory
- self._manager_finders[class_] = manager.manager_getter()
- self._state_finders[class_] = manager.state_getter()
- self._dict_finders[class_] = manager.dict_getter()
-
- self.dispatch.class_instrument(class_)
-
- return manager
-
- def _collect_management_factories_for(self, cls):
- """Return a collection of factories in play or specified for a
- hierarchy.
-
- Traverses the entire inheritance graph of a cls and returns a
- collection of instrumentation factories for those classes. Factories
- are extracted from active ClassManagers, if available, otherwise
- instrumentation_finders is consulted.
-
- """
- hierarchy = util.class_hierarchy(cls)
- factories = set()
- for member in hierarchy:
- manager = manager_of_class(member)
- if manager is not None:
- factories.add(manager.factory)
- else:
- for finder in instrumentation_finders:
- factory = finder(member)
- if factory is not None:
- break
- else:
- factory = None
- factories.add(factory)
- factories.discard(None)
- return factories
-
- def manager_of_class(self, cls):
- # this is only called when alternate instrumentation
- # has been established
- if cls is None:
- return None
- try:
- finder = self._manager_finders[cls]
- except KeyError:
- return None
- else:
- return finder(cls)
-
- def state_of(self, instance):
- # this is only called when alternate instrumentation
- # has been established
- if instance is None:
- raise AttributeError("None has no persistent state.")
- try:
- return self._state_finders[instance.__class__](instance)
- except KeyError:
- raise AttributeError("%r is not instrumented" %
- instance.__class__)
-
- def dict_of(self, instance):
- # this is only called when alternate instrumentation
- # has been established
- if instance is None:
- raise AttributeError("None has no persistent state.")
- try:
- return self._dict_finders[instance.__class__](instance)
- except KeyError:
- raise AttributeError("%r is not instrumented" %
- instance.__class__)
-
- def unregister(self, class_):
- if class_ in self._manager_finders:
- manager = self.manager_of_class(class_)
- self.dispatch.class_uninstrument(class_)
- manager.unregister()
- manager.dispose()
- del self._manager_finders[class_]
- del self._state_finders[class_]
- del self._dict_finders[class_]
- if ClassManager.MANAGER_ATTR in class_.__dict__:
- delattr(class_, ClassManager.MANAGER_ATTR)
+# 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 = ClassManager.state_getter()
-instrumentation_registry = InstrumentationRegistry()
+instance_dict = _default_dict_getter = ClassManager.dict_getter()
-
-def _install_lookup_strategy(implementation):
- """Replace global class/object management functions
- with either faster or more comprehensive implementations,
- based on whether or not extended class instrumentation
- has been detected.
-
- This function is called only by InstrumentationRegistry()
- and unit tests specific to this behavior.
-
- """
- global instance_state, instance_dict, manager_of_class
- if implementation is util.symbol('native'):
- instance_state = attrgetter(ClassManager.STATE_ATTR)
- instance_dict = attrgetter("__dict__")
- def manager_of_class(cls):
- return cls.__dict__.get(ClassManager.MANAGER_ATTR, None)
- else:
- instance_state = instrumentation_registry.state_of
- instance_dict = instrumentation_registry.dict_of
- manager_of_class = instrumentation_registry.manager_of_class
- attributes.instance_state = instance_state
- attributes.instance_dict = instance_dict
- attributes.manager_of_class = manager_of_class
-
-_create_manager_for_cls = instrumentation_registry.create_manager_for_cls
-
-# Install default "lookup" strategies. These are basically
-# very fast attrgetters for key attributes.
-# When a custom ClassManager is installed, more expensive per-class
-# strategies are copied over these.
-_install_lookup_strategy(util.symbol('native'))
-
-
-def find_native_user_instrumentation_hook(cls):
- """Find user-specified instrumentation management for a class."""
- return getattr(cls, INSTRUMENTATION_MANAGER, None)
-instrumentation_finders.append(find_native_user_instrumentation_hook)
+manager_of_class = _default_manager_getter = ClassManager.manager_getter()
def _generate_init(class_, class_manager):
"""Build an __init__ decorator that triggers ClassManager events."""
#if func_kw_defaults:
# __init__.__kwdefaults__ = func_kw_defaults
return __init__
+
from .. import exc as sa_exc, util
from ..sql import operators
from collections import deque
+#from . import _instrumentation_ext
+#InstrumentationManager = _instrumentation_ext.InstrumentationManager
+#from ..ext.instrumentation import InstrumentationManager
orm_util = util.importlater('sqlalchemy.orm', 'util')
collections = util.importlater('sqlalchemy.orm', 'collections')
return str(self.parent_property)
-class InstrumentationManager(object):
- """User-defined class instrumentation extension.
-
- :class:`.InstrumentationManager` can be subclassed in order
- to change
- how class instrumentation proceeds. This class exists for
- the purposes of integration with other object management
- frameworks which would like to entirely modify the
- instrumentation methodology of the ORM, and is not intended
- for regular usage. For interception of class instrumentation
- events, see :class:`.InstrumentationEvents`.
-
- For an example of :class:`.InstrumentationManager`, see the
- example :ref:`examples_instrumentation`.
-
- The API for this class should be considered as semi-stable,
- and may change slightly with new releases.
-
- """
-
- # r4361 added a mandatory (cls) constructor to this interface.
- # given that, perhaps class_ should be dropped from all of these
- # signatures.
-
- def __init__(self, class_):
- pass
-
- def manage(self, class_, manager):
- setattr(class_, '_default_class_manager', manager)
-
- def dispose(self, class_, manager):
- delattr(class_, '_default_class_manager')
-
- def manager_getter(self, class_):
- def get(cls):
- return cls._default_class_manager
- return get
-
- def instrument_attribute(self, class_, key, inst):
- pass
-
- def post_configure_attribute(self, class_, key, inst):
- pass
-
- def install_descriptor(self, class_, key, inst):
- setattr(class_, key, inst)
-
- def uninstall_descriptor(self, class_, key):
- delattr(class_, key)
-
- def install_member(self, class_, key, implementation):
- setattr(class_, key, implementation)
-
- def uninstall_member(self, class_, key):
- delattr(class_, key)
-
- def instrument_collection_class(self, class_, key, collection_class):
- return collections.prepare_instrumentation(collection_class)
-
- def get_instance_dict(self, class_, instance):
- return instance.__dict__
-
- def initialize_instance_dict(self, class_, instance):
- pass
-
- def install_state(self, class_, instance, state):
- setattr(instance, '_default_state', state)
-
- def remove_state(self, class_, instance):
- delattr(instance, '_default_state')
-
- def state_getter(self, class_):
- return lambda instance: getattr(instance, '_default_state')
-
- def dict_getter(self, class_):
- return lambda inst: self.get_instance_dict(class_, inst)
SQL_OK, NEVER_SET, ATTR_WAS_SET, NO_VALUE,\
PASSIVE_NO_INITIALIZE
)
-mapperlib = util.importlater("sqlalchemy.orm", "mapperlib")
-sessionlib = util.importlater("sqlalchemy.orm", "session")
+from . import mapperlib
+from . import session as sessionlib
class InstanceState(object):
OrderedSet, IdentitySet, OrderedIdentitySet, column_set, \
column_dict, ordered_column_set, populate_column_dict, unique_list, \
UniqueAppender, PopulateDict, EMPTY_SET, to_list, to_set, \
- to_column_set, update_copy, flatten_iterator, WeakIdentityMapping, \
+ to_column_set, update_copy, flatten_iterator, \
LRUCache, ScopedRegistry, ThreadLocalRegistry
from langhelpers import iterate_attributes, class_hierarchy, \
duck_type_collection, assert_arg_type, symbol, dictlike_iteritems,\
classproperty, set_creation_order, warn_exception, warn, NoneType,\
constructor_copy, methods_equivalent, chop_traceback, asint,\
- generic_repr, counter, PluginLoader
+ generic_repr, counter, PluginLoader, hybridmethod
from deprecations import warn_deprecated, warn_pending_deprecation, \
deprecated, pending_deprecation
else:
yield elem
-class WeakIdentityMapping(weakref.WeakKeyDictionary):
- """A WeakKeyDictionary with an object identity index.
-
- Adds a .by_id dictionary to a regular WeakKeyDictionary. Trades
- performance during mutation operations for accelerated lookups by id().
-
- The usual cautions about weak dictionaries and iteration also apply to
- this subclass.
-
- """
- _none = symbol('none')
-
- def __init__(self):
- weakref.WeakKeyDictionary.__init__(self)
- self.by_id = {}
- self._weakrefs = {}
-
- def __setitem__(self, object, value):
- oid = id(object)
- self.by_id[oid] = value
- if oid not in self._weakrefs:
- self._weakrefs[oid] = self._ref(object)
- weakref.WeakKeyDictionary.__setitem__(self, object, value)
-
- def __delitem__(self, object):
- del self._weakrefs[id(object)]
- del self.by_id[id(object)]
- weakref.WeakKeyDictionary.__delitem__(self, object)
-
- def setdefault(self, object, default=None):
- value = weakref.WeakKeyDictionary.setdefault(self, object, default)
- oid = id(object)
- if value is default:
- self.by_id[oid] = default
- if oid not in self._weakrefs:
- self._weakrefs[oid] = self._ref(object)
- return value
-
- def pop(self, object, default=_none):
- if default is self._none:
- value = weakref.WeakKeyDictionary.pop(self, object)
- else:
- value = weakref.WeakKeyDictionary.pop(self, object, default)
- if id(object) in self.by_id:
- del self._weakrefs[id(object)]
- del self.by_id[id(object)]
- return value
-
- def popitem(self):
- item = weakref.WeakKeyDictionary.popitem(self)
- oid = id(item[0])
- del self._weakrefs[oid]
- del self.by_id[oid]
- return item
-
- def clear(self):
- # Py2K
- # in 3k, MutableMapping calls popitem()
- self._weakrefs.clear()
- self.by_id.clear()
- # end Py2K
- weakref.WeakKeyDictionary.clear(self)
-
- def update(self, *a, **kw):
- raise NotImplementedError
-
- def _cleanup(self, wr, key=None):
- if key is None:
- key = wr.key
- try:
- del self._weakrefs[key]
- except (KeyError, AttributeError): # pragma: no cover
- pass # pragma: no cover
- try:
- del self.by_id[key]
- except (KeyError, AttributeError): # pragma: no cover
- pass # pragma: no cover
-
- class _keyed_weakref(weakref.ref):
- def __init__(self, object, callback):
- weakref.ref.__init__(self, object, callback)
- self.key = id(object)
-
- def _ref(self, object):
- return self._keyed_weakref(object, self._cleanup)
class LRUCache(dict):
def module(self):
if self in importlater._unresolved:
raise ImportError(
- "importlater.resolve_all() hasn't been called")
+ "importlater.resolve_all() hasn't "
+ "been called (this is %s %s)"
+ % (self._il_path, self._il_addtl))
m = self._initial_import
if self._il_addtl:
def __get__(desc, self, cls):
return desc.fget(cls)
+class hybridmethod(object):
+ """Decorate a function as cls- or instance- level."""
+ def __init__(self, func, expr=None):
+ self.func = func
+
+ def __get__(self, instance, owner):
+ if instance is None:
+ return self.func.__get__(owner, owner.__class__)
+ else:
+ return self.func.__get__(instance, owner)
+
class _symbol(int):
def __new__(self, name, doc=None, canonical=None):
-from test.lib.testing import eq_, assert_raises, assert_raises_message
-import pickle
+from test.lib.testing import eq_, assert_raises, assert_raises_message, ne_
from sqlalchemy import util
-from sqlalchemy.orm import attributes, instrumentation
-from sqlalchemy.orm.collections import collection
+from sqlalchemy.orm import attributes
from sqlalchemy.orm.attributes import set_attribute, get_attribute, del_attribute
from sqlalchemy.orm.instrumentation import is_instrumented
from sqlalchemy.orm import clear_mappers
-from sqlalchemy.orm import InstrumentationManager
from test.lib import *
from test.lib import fixtures
-
-class MyTypesManager(InstrumentationManager):
+from sqlalchemy.ext import instrumentation
+from sqlalchemy.orm.instrumentation import register_class
+from test.lib.util import decorator
+from sqlalchemy.orm import events
+from sqlalchemy import event
+
+@decorator
+def modifies_instrumentation_finders(fn, *args, **kw):
+ pristine = instrumentation.instrumentation_finders[:]
+ try:
+ fn(*args, **kw)
+ finally:
+ del instrumentation.instrumentation_finders[:]
+ instrumentation.instrumentation_finders.extend(pristine)
+
+def with_lookup_strategy(strategy):
+ @decorator
+ def decorate(fn, *args, **kw):
+ try:
+ ext_instrumentation._install_instrumented_lookups()
+ return fn(*args, **kw)
+ finally:
+ ext_instrumentation._reinstall_default_lookups()
+ return decorate
+
+
+class MyTypesManager(instrumentation.InstrumentationManager):
def instrument_attribute(self, class_, key, attr):
pass
remove = _sa_remover
class MyBaseClass(object):
- __sa_instrumentation_manager__ = InstrumentationManager
+ __sa_instrumentation_manager__ = instrumentation.InstrumentationManager
class MyClass(object):
@classmethod
def teardown_class(cls):
clear_mappers()
- instrumentation._install_lookup_strategy(util.symbol('native'))
+ instrumentation._reinstall_default_lookups()
def test_instance_dict(self):
class User(MyClass):
pass
- instrumentation.register_class(User)
+ register_class(User)
attributes.register_attribute(User, 'user_id', uselist = False, useobject=False)
attributes.register_attribute(User, 'user_name', uselist = False, useobject=False)
attributes.register_attribute(User, 'email_address', uselist = False, useobject=False)
class User(base):
pass
- instrumentation.register_class(User)
+ register_class(User)
attributes.register_attribute(User, 'user_id', uselist = False, useobject=False)
attributes.register_attribute(User, 'user_name', uselist = False, useobject=False)
attributes.register_attribute(User, 'email_address', uselist = False, useobject=False)
def test_deferred(self):
for base in (object, MyBaseClass, MyClass):
- class Foo(base):pass
+ class Foo(base):
+ pass
data = {'a':'this is a', 'b':12}
def loader(state, keys):
state.dict[k] = data[k]
return attributes.ATTR_WAS_SET
- manager = instrumentation.register_class(Foo)
+ manager = register_class(Foo)
manager.deferred_scalar_loader = loader
attributes.register_attribute(Foo, 'a', uselist=False, useobject=False)
attributes.register_attribute(Foo, 'b', uselist=False, useobject=False)
- assert Foo in instrumentation.instrumentation_registry._state_finders
+ if base is object:
+ assert Foo not in instrumentation._instrumentation_factory._state_finders
+ else:
+ assert Foo in instrumentation._instrumentation_factory._state_finders
+
f = Foo()
attributes.instance_state(f).expire(attributes.instance_dict(f), set())
eq_(f.a, "this is a")
class Foo(base):pass
class Bar(Foo):pass
- instrumentation.register_class(Foo)
- instrumentation.register_class(Bar)
+ register_class(Foo)
+ register_class(Bar)
def func1(state, passive):
return "this is the foo attr"
class Post(base):pass
class Blog(base):pass
- instrumentation.register_class(Post)
- instrumentation.register_class(Blog)
+ register_class(Post)
+ register_class(Blog)
attributes.register_attribute(Post, 'blog', uselist=False,
backref='posts', trackparent=True, useobject=True)
attributes.register_attribute(Blog, 'posts', uselist=True,
class Bar(base):
pass
- instrumentation.register_class(Foo)
- instrumentation.register_class(Bar)
+ register_class(Foo)
+ register_class(Bar)
attributes.register_attribute(Foo, "name", uselist=False, useobject=False)
attributes.register_attribute(Foo, "bars", uselist=True, trackparent=True, useobject=True)
attributes.register_attribute(Bar, "name", uselist=False, useobject=False)
def test_null_instrumentation(self):
class Foo(MyBaseClass):
pass
- instrumentation.register_class(Foo)
+ register_class(Foo)
attributes.register_attribute(Foo, "name", uselist=False, useobject=False)
attributes.register_attribute(Foo, "bars", uselist=True, trackparent=True, useobject=True)
class Unknown(object): pass
class Known(MyBaseClass): pass
- instrumentation.register_class(Known)
+ register_class(Known)
k, u = Known(), Unknown()
assert instrumentation.manager_of_class(Unknown) is None
attributes.instance_state, None)
-if __name__ == '__main__':
- testing.main()
+class FinderTest(fixtures.ORMTest):
+ def test_standard(self):
+ class A(object): pass
+
+ register_class(A)
+
+ eq_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
+
+ def test_nativeext_interfaceexact(self):
+ class A(object):
+ __sa_instrumentation_manager__ = instrumentation.InstrumentationManager
+
+ register_class(A)
+ ne_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
+
+ def test_nativeext_submanager(self):
+ class Mine(instrumentation.ClassManager): pass
+ class A(object):
+ __sa_instrumentation_manager__ = Mine
+
+ register_class(A)
+ eq_(type(instrumentation.manager_of_class(A)), Mine)
+
+ @modifies_instrumentation_finders
+ def test_customfinder_greedy(self):
+ class Mine(instrumentation.ClassManager): pass
+ class A(object): pass
+ def find(cls):
+ return Mine
+
+ instrumentation.instrumentation_finders.insert(0, find)
+ register_class(A)
+ eq_(type(instrumentation.manager_of_class(A)), Mine)
+
+ @modifies_instrumentation_finders
+ def test_customfinder_pass(self):
+ class A(object): pass
+ def find(cls):
+ return None
+
+ instrumentation.instrumentation_finders.insert(0, find)
+ register_class(A)
+ eq_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
+
+class InstrumentationCollisionTest(fixtures.ORMTest):
+ def test_none(self):
+ class A(object): pass
+ register_class(A)
+
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+ class B(object):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+ register_class(B)
+
+ class C(object):
+ __sa_instrumentation_manager__ = instrumentation.ClassManager
+ register_class(C)
+
+ def test_single_down(self):
+ class A(object): pass
+ register_class(A)
+
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+ class B(A):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+
+ assert_raises_message(TypeError, "multiple instrumentation implementations", register_class, B)
+
+ def test_single_up(self):
+
+ class A(object): pass
+ # delay registration
+
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+ class B(A):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+ register_class(B)
+
+ assert_raises_message(TypeError, "multiple instrumentation implementations", register_class, A)
+
+ def test_diamond_b1(self):
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+
+ class A(object): pass
+ class B1(A): pass
+ class B2(A):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+ class C(object): pass
+
+ assert_raises_message(TypeError, "multiple instrumentation implementations", register_class, B1)
+
+ def test_diamond_b2(self):
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+
+ class A(object): pass
+ class B1(A): pass
+ class B2(A):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+ class C(object): pass
+
+ register_class(B2)
+ assert_raises_message(TypeError, "multiple instrumentation implementations", register_class, B1)
+
+ def test_diamond_c_b(self):
+ mgr_factory = lambda cls: instrumentation.ClassManager(cls)
+
+ class A(object): pass
+ class B1(A): pass
+ class B2(A):
+ __sa_instrumentation_manager__ = staticmethod(mgr_factory)
+ class C(object): pass
+
+ register_class(C)
+
+ assert_raises_message(TypeError, "multiple instrumentation implementations", register_class, B1)
+
+
+class ExtendedEventsTest(fixtures.ORMTest):
+ """Allow custom Events implementations."""
+
+ @modifies_instrumentation_finders
+ def test_subclassed(self):
+ class MyEvents(events.InstanceEvents):
+ pass
+ class MyClassManager(instrumentation.ClassManager):
+ dispatch = event.dispatcher(MyEvents)
+
+ instrumentation.instrumentation_finders.insert(0, lambda cls: MyClassManager)
+
+ class A(object): pass
+
+ register_class(A)
+ manager = instrumentation.manager_of_class(A)
+ assert issubclass(manager.dispatch._parent_cls.__dict__['dispatch'].events, MyEvents)
+
from test.lib.schema import Table
from test.lib.schema import Column
from test.lib.testing import eq_, ne_
-from test.lib.util import decorator
from test.lib import fixtures, testing
-@decorator
-def modifies_instrumentation_finders(fn, *args, **kw):
- pristine = instrumentation.instrumentation_finders[:]
- try:
- fn(*args, **kw)
- finally:
- del instrumentation.instrumentation_finders[:]
- instrumentation.instrumentation_finders.extend(pristine)
-
-def with_lookup_strategy(strategy):
- @decorator
- def decorate(fn, *args, **kw):
- try:
- instrumentation._install_lookup_strategy(strategy)
- return fn(*args, **kw)
- finally:
- instrumentation._install_lookup_strategy(sa.util.symbol('native'))
- return decorate
-
class InitTest(fixtures.ORMTest):
def fixture(self):
# C is not mapped in the current implementation
assert_raises(sa.orm.exc.UnmappedClassError, class_mapper, C)
-class InstrumentationCollisionTest(fixtures.ORMTest):
- def test_none(self):
- class A(object): pass
- instrumentation.register_class(A)
-
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
- class B(object):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
- instrumentation.register_class(B)
-
- class C(object):
- __sa_instrumentation_manager__ = instrumentation.ClassManager
- instrumentation.register_class(C)
-
- def test_single_down(self):
- class A(object): pass
- instrumentation.register_class(A)
-
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
- class B(A):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
-
- assert_raises_message(TypeError, "multiple instrumentation implementations", instrumentation.register_class, B)
-
- def test_single_up(self):
-
- class A(object): pass
- # delay registration
-
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
- class B(A):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
- instrumentation.register_class(B)
-
- assert_raises_message(TypeError, "multiple instrumentation implementations", instrumentation.register_class, A)
-
- def test_diamond_b1(self):
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
-
- class A(object): pass
- class B1(A): pass
- class B2(A):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object): pass
-
- assert_raises_message(TypeError, "multiple instrumentation implementations", instrumentation.register_class, B1)
-
- def test_diamond_b2(self):
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
-
- class A(object): pass
- class B1(A): pass
- class B2(A):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object): pass
-
- instrumentation.register_class(B2)
- assert_raises_message(TypeError, "multiple instrumentation implementations", instrumentation.register_class, B1)
-
- def test_diamond_c_b(self):
- mgr_factory = lambda cls: instrumentation.ClassManager(cls)
-
- class A(object): pass
- class B1(A): pass
- class B2(A):
- __sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object): pass
-
- instrumentation.register_class(C)
-
- assert_raises_message(TypeError, "multiple instrumentation implementations", instrumentation.register_class, B1)
class OnLoadTest(fixtures.ORMTest):
"""Check that Events.load is not hit in regular attributes operations."""
finally:
del A
- @classmethod
- def teardown_class(cls):
- clear_mappers()
- instrumentation._install_lookup_strategy(util.symbol('native'))
-
-
-class ExtendedEventsTest(fixtures.ORMTest):
- """Allow custom Events implementations."""
-
- @modifies_instrumentation_finders
- def test_subclassed(self):
- class MyEvents(events.InstanceEvents):
- pass
- class MyClassManager(instrumentation.ClassManager):
- dispatch = event.dispatcher(MyEvents)
-
- instrumentation.instrumentation_finders.insert(0, lambda cls: MyClassManager)
-
- class A(object): pass
-
- instrumentation.register_class(A)
- manager = instrumentation.manager_of_class(A)
- assert issubclass(manager.dispatch._parent_cls.__dict__['dispatch'].events, MyEvents)
-
class NativeInstrumentationTest(fixtures.ORMTest):
- @with_lookup_strategy(sa.util.symbol('native'))
def test_register_reserved_attribute(self):
class T(object): pass
fails('install_descriptor', sa)
fails('install_descriptor', ma)
- @with_lookup_strategy(sa.util.symbol('native'))
def test_mapped_stateattr(self):
t = Table('t', MetaData(),
Column('id', Integer, primary_key=True),
assert_raises(KeyError, mapper, T, t)
- @with_lookup_strategy(sa.util.symbol('native'))
def test_mapped_managerattr(self):
t = Table('t', MetaData(),
Column('id', Integer, primary_key=True),
class A(object):pass
manager = instrumentation.register_class(A)
+ attributes.register_attribute(A, 'x', uselist=False, useobject=False)
assert instrumentation.manager_of_class(A) is manager
instrumentation.unregister_class(A)
assert instrumentation.manager_of_class(A) is None
+ assert not hasattr(A, 'x')
+ assert A.__init__ is object.__init__
def test_compileonattr_rel_backref_a(self):
m = MetaData()
assert b in session, 'base: %s' % base
-class FinderTest(fixtures.ORMTest):
- def test_standard(self):
- class A(object): pass
-
- instrumentation.register_class(A)
-
- eq_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
-
- def test_nativeext_interfaceexact(self):
- class A(object):
- __sa_instrumentation_manager__ = sa.orm.interfaces.InstrumentationManager
-
- instrumentation.register_class(A)
- ne_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
-
- def test_nativeext_submanager(self):
- class Mine(instrumentation.ClassManager): pass
- class A(object):
- __sa_instrumentation_manager__ = Mine
-
- instrumentation.register_class(A)
- eq_(type(instrumentation.manager_of_class(A)), Mine)
-
- @modifies_instrumentation_finders
- def test_customfinder_greedy(self):
- class Mine(instrumentation.ClassManager): pass
- class A(object): pass
- def find(cls):
- return Mine
-
- instrumentation.instrumentation_finders.insert(0, find)
- instrumentation.register_class(A)
- eq_(type(instrumentation.manager_of_class(A)), Mine)
-
- @modifies_instrumentation_finders
- def test_customfinder_pass(self):
- class A(object): pass
- def find(cls):
- return None
-
- instrumentation.instrumentation_finders.insert(0, find)
- instrumentation.register_class(A)
- eq_(type(instrumentation.manager_of_class(A)), instrumentation.ClassManager)
-
-
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
