1. Daniel Miller
  2. sqlalchemy

Commits

Mike Bayer  committed 4dbd52a

- Fixed regression from 0.6 where a get history
operation on some relationship() based attributes
would fail when a lazyload would emit; this could
trigger within a flush() under certain conditions.
[ticket:2224] Thanks to the user who submitted
the great test for this.

  • Participants
  • Parent commits 9888bda
  • Branches default

Comments (0)

Files changed (5)

File CHANGES

View file
     _with_invoke_all_eagers()
     which selects old/new behavior [ticket:2213]
 
+  - Fixed regression from 0.6 where a get history
+    operation on some relationship() based attributes
+    would fail when a lazyload would emit; this could 
+    trigger within a flush() under certain conditions.
+    [ticket:2224]  Thanks to the user who submitted
+    the great test for this.
+
   - Fixed bug whereby the source clause
     used by query.join() would be inconsistent
     if against a column expression that combined

File lib/sqlalchemy/orm/attributes.py

View file
 NEVER_SET = util.symbol('NEVER_SET')
 
 PASSIVE_RETURN_NEVER_SET = util.symbol('PASSIVE_RETURN_NEVER_SET'
-"""Symbol indicating that a 'default' value, i.e. None or blank
-collection, should not be assigned to an attribute when a get()
-is performed and no value was present.  NEVER_SET is returned
-instead.
+"""Symbol indicating that loader callables can be 
+fired off, but if no callable is applicable and no value is
+present, the attribute should remain non-initialized.
+NEVER_SET is returned in this case.
 """)
 
 PASSIVE_NO_INITIALIZE = util.symbol('PASSIVE_NO_INITIALIZE',
                 else:
                     value = ATTR_EMPTY
 
-                if value is PASSIVE_NO_RESULT:
+                if value in (PASSIVE_NO_RESULT, NEVER_SET):
                     return value
                 elif value is ATTR_WAS_SET:
                     try:

File lib/sqlalchemy/orm/strategies.py

View file
             ]
             if attributes.PASSIVE_NO_RESULT in ident:
                 return attributes.PASSIVE_NO_RESULT
+            elif attributes.NEVER_SET in ident:
+                return attributes.NEVER_SET
 
             if _none_set.issuperset(ident):
                 return None

File test/orm/test_attributes.py

View file
             ('remove', f, [4,5,6])
         ])
 
-
     def test_lazytrackparent(self):
         """test that the "hasparent" flag works properly 
            when lazy loaders and backrefs are used
         except sa_exc.ArgumentError, e:
             assert False
 
+class GetNoValueTest(fixtures.ORMTest):
+    def _fixture(self, expected):
+        class Foo(object):
+            pass
+
+        class Bar(object):
+            pass
+
+        def lazy_callable(state, passive):
+            return expected
+
+        instrumentation.register_class(Foo)
+        instrumentation.register_class(Bar)
+        if expected is not None:
+            attributes.register_attribute(Foo, 
+                        "attr", useobject=True, 
+                        uselist=False, callable_=lazy_callable)
+        else:
+            attributes.register_attribute(Foo, 
+                        "attr", useobject=True, 
+                        uselist=False)
+
+        f1 = Foo()
+        return Foo.attr.impl,\
+                attributes.instance_state(f1), \
+                attributes.instance_dict(f1)
+
+
+    def test_passive_no_result(self):
+        attr, state, dict_ = self._fixture(attributes.PASSIVE_NO_RESULT)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.PASSIVE_NO_RESULT
+        )
+
+    def test_passive_no_result_never_set(self):
+        attr, state, dict_ = self._fixture(attributes.NEVER_SET)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.PASSIVE_NO_RESULT
+        )
+        assert 'attr' not in dict_
+
+    def test_passive_ret_never_set_never_set(self):
+        attr, state, dict_ = self._fixture(attributes.NEVER_SET)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            attributes.NEVER_SET
+        )
+        assert 'attr' not in dict_
+
+    def test_passive_ret_never_set_empty(self):
+        attr, state, dict_ = self._fixture(None)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            attributes.NEVER_SET
+        )
+        assert 'attr' not in dict_
+
+    def test_passive_no_result_empty(self):
+        attr, state, dict_ = self._fixture(None)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_NO_RESULT),
+            None
+        )
+        assert 'attr' in dict_
+
+    def test_off_empty(self):
+        attr, state, dict_ = self._fixture(None)
+        eq_(
+            attr.get(state, dict_, passive=attributes.PASSIVE_OFF),
+            None
+        )
+        assert 'attr' in dict_
+
 class UtilTest(fixtures.ORMTest):
     def test_helpers(self):
         class Foo(object):

File test/orm/test_lazy_relations.py

View file
 
         assert a.user is u1
 
+
+
     def test_backrefs_dont_lazyload(self):
         users, Address, addresses, User = (self.tables.users,
                                 self.classes.Address,
             assert ad2 in u1.addresses
         self.assert_sql_count(testing.db, go, 1)
 
+class GetterStateTest(_fixtures.FixtureTest):
+    """test lazyloader on non-existent attribute returns 
+    expected attribute symbols, maintain expected state"""
+
+    run_inserts = None
+
+    def _u_ad_fixture(self, populate_user):
+        users, Address, addresses, User = (self.tables.users,
+                                self.classes.Address,
+                                self.tables.addresses,
+                                self.classes.User)
+
+        mapper(User, users, properties={
+            'addresses':relationship(Address, backref='user')
+        })
+        mapper(Address, addresses)
+
+        sess = create_session()
+        a1 = Address(email_address='a1')
+        sess.add(a1)
+        if populate_user:
+            a1.user = User(name='ed')
+        sess.flush()
+        if populate_user:
+            sess.expire_all()
+        return User, Address, sess, a1
+
+    def test_get_empty_passive_return_never_set(self):
+        User, Address, sess, a1 = self._u_ad_fixture(False)
+        eq_(
+            Address.user.impl.get(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            attributes.NEVER_SET
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_history_empty_passive_return_never_set(self):
+        User, Address, sess, a1 = self._u_ad_fixture(False)
+        eq_(
+            Address.user.impl.get_history(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            ((), (), ())
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_get_empty_passive_no_initialize(self):
+        User, Address, sess, a1 = self._u_ad_fixture(False)
+        eq_(
+            Address.user.impl.get(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.PASSIVE_NO_RESULT
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_history_empty_passive_no_initialize(self):
+        User, Address, sess, a1 = self._u_ad_fixture(False)
+        eq_(
+            Address.user.impl.get_history(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.HISTORY_BLANK
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_get_populated_passive_no_initialize(self):
+        User, Address, sess, a1 = self._u_ad_fixture(True)
+        eq_(
+            Address.user.impl.get(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.PASSIVE_NO_RESULT
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_history_populated_passive_no_initialize(self):
+        User, Address, sess, a1 = self._u_ad_fixture(True)
+        eq_(
+            Address.user.impl.get_history(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_NO_INITIALIZE),
+            attributes.HISTORY_BLANK
+        )
+        assert 'user_id' not in a1.__dict__
+        assert 'user' not in a1.__dict__
+
+    def test_get_populated_passive_return_never_set(self):
+        User, Address, sess, a1 = self._u_ad_fixture(True)
+        eq_(
+            Address.user.impl.get(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            User(name='ed')
+        )
+
+    def test_history_populated_passive_return_never_set(self):
+        User, Address, sess, a1 = self._u_ad_fixture(True)
+        eq_(
+            Address.user.impl.get_history(
+                attributes.instance_state(a1), 
+                attributes.instance_dict(a1), 
+                passive=attributes.PASSIVE_RETURN_NEVER_SET),
+            ((), [User(name='ed'), ], ())
+        )
 
 class M2OGetTest(_fixtures.FixtureTest):
     run_inserts = 'once'