Ensure mapping has no version_id_generator when checking missing version_id

Fixed 1.2 regression in ORM versioning feature where a mapping against a
:func:`.select` or :func:`.alias` that also used a versioning column
against the underlying table would fail due to the check added as part of
:ticket:`3673`.

See also #4194 and #4195 for related issues found regarding
mapped selects and versioning.

Change-Id: Ifabe9a5c8c1bfa72db5029faa7a5dbf71f0a7ca0
Fixes: #4193

diff --git a/doc/build/changelog/unreleased_12/4193.rst b/doc/build/changelog/unreleased_12/4193.rst
new file mode 100644 (file)
index 0000000..3afc2f7
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4193.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4193
+
+    Fixed 1.2 regression in ORM versioning feature where a mapping against a
+    :func:`.select` or :func:`.alias` that also used a versioning column
+    against the underlying table would fail due to the check added as part of
+    :ticket:`3673`.

diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index dc0ae1c38a5bccb27a57662f564f7c75261e0442..4f1e08afacd20b40489b24ab1e068777b7bb64ff 100644 (file)
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1127,7 +1127,8 @@ def _finalize_insert_update_commands(base_mapper, uowtransaction, states):
         else:
             mapper.dispatch.after_update(mapper, connection, state)
 
-        if mapper.version_id_col is not None:
+        if mapper.version_id_generator is False and \
+                mapper.version_id_col is not None:
             if state_dict[mapper._version_id_prop.key] is None:
                 raise orm_exc.FlushError(
                     "Instance does not contain a non-NULL version value")

diff --git a/test/orm/test_versioning.py b/test/orm/test_versioning.py
index 089541848b9d1d1564b1edc339eda93877f726b1..a1eef5d1646ae78603c9804dddbabebc31eb44b7 100644 (file)
--- a/test/orm/test_versioning.py
+++ b/test/orm/test_versioning.py
@@ -1642,3 +1642,86 @@ class ManualInheritanceVersionTest(fixtures.MappedTest):
         sess.commit()
 
         eq_(b1.vid, 2)
+
+
+class VersioningMappedSelectTest(fixtures.MappedTest):
+    # test for #4193, see also #4194 for related notes
+
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('version_table', metadata,
+              Column('id', Integer, primary_key=True,
+                     test_needs_autoincrement=True),
+              Column('version_id', Integer, nullable=False),
+              Column('value', String(40), nullable=False))
+
+    @classmethod
+    def setup_classes(cls):
+        class Foo(cls.Basic):
+            pass
+
+    def _implicit_version_fixture(self):
+        Foo, version_table = self.classes.Foo, self.tables.version_table
+
+        current = version_table.select().\
+            where(version_table.c.id > 0).alias('current_table')
+
+        mapper(Foo, current, version_id_col=version_table.c.version_id)
+        s1 = Session()
+        return s1
+
+    def _explicit_version_fixture(self):
+        Foo, version_table = self.classes.Foo, self.tables.version_table
+
+        current = version_table.select().\
+            where(version_table.c.id > 0).alias('current_table')
+
+        mapper(Foo, current,
+               version_id_col=version_table.c.version_id,
+               version_id_generator=False)
+        s1 = Session()
+        return s1
+
+    @testing.emits_warning(r".*versioning cannot be verified")
+    def test_implicit(self):
+        Foo = self.classes.Foo
+
+        s1 = self._implicit_version_fixture()
+        f1 = Foo(value='f1')
+        f2 = Foo(value='f2')
+        s1.add_all((f1, f2))
+        s1.commit()
+
+        f1.value = 'f1rev2'
+        f2.value = 'f2rev2'
+        s1.commit()
+
+        eq_(
+            s1.query(Foo.id, Foo.value, Foo.version_id).order_by(Foo.id).all(),
+            [(f1.id, 'f1rev2', 2), (f2.id, 'f2rev2', 2)]
+        )
+
+    @testing.emits_warning(r".*versioning cannot be verified")
+    def test_explicit(self):
+        Foo = self.classes.Foo
+
+        s1 = self._explicit_version_fixture()
+        f1 = Foo(value='f1', version_id=1)
+        f2 = Foo(value='f2', version_id=1)
+        s1.add_all((f1, f2))
+        s1.flush()
+
+        # note this requires that the Session was not expired until
+        # we fix #4195
+        f1.value = 'f1rev2'
+        f1.version_id = 2
+        f2.value = 'f2rev2'
+        f2.version_id = 2
+        s1.flush()
+
+        eq_(
+            s1.query(Foo.id, Foo.value, Foo.version_id).order_by(Foo.id).all(),
+            [(f1.id, 'f1rev2', 2), (f2.id, 'f2rev2', 2)]
+        )
\ No newline at end of file