setattr(instance, self.STATE_ATTR, state or self.instance_state_factory(instance, self))
return instance
+ def setup_instance(self, instance, state=None):
+ setattr(instance, self.STATE_ATTR, state or self.instance_state_factory(instance, self))
+
def _new_state_if_none(self, instance):
"""Install a default InstanceState if none is present.
def new_instance(self, state=None):
instance = self.class_.__new__(self.class_)
- self._setup_instance(instance, state)
+ self.setup_instance(instance, state)
return instance
def _new_state_if_none(self, instance):
if self.has_state(instance):
return False
else:
- return self._setup_instance(instance)
+ return self.setup_instance(instance)
- def _setup_instance(self, instance, state=None):
+ def setup_instance(self, instance, state=None):
self._adapted.initialize_instance_dict(self.class_, instance)
if state is None:
if instance is EXT_CONTINUE:
instance = self.class_manager.new_instance()
else:
- # TODO: don't think theres coverage here
manager = attributes.manager_of_class(instance.__class__)
# TODO: if manager is None, raise a friendly error about
# returning instances of unmapped types
'create_instance', 'populate_instance', 'reconstruct_instance',
'append_result', 'before_update', 'after_update', 'before_delete',
'after_delete'])
-
+
+ @testing.resolve_artifact_names
+ def test_create_instance(self):
+ class CreateUserExt(sa.orm.MapperExtension):
+ def create_instance(self, mapper, selectcontext, row, class_):
+ return User.__new__(User)
+
+ mapper(User, users, extension=CreateUserExt())
+ sess = create_session()
+ u1 = User()
+ u1.name = 'ed'
+ sess.add(u1)
+ sess.flush()
+ sess.clear()
+ assert sess.query(User).first()
+
class RequirementsTest(_base.MappedTest):
"""Tests the contract for user classes."""