diff options
| -rw-r--r-- | doc/build/changelog/changelog_11.rst | 9 | ||||
| -rw-r--r-- | lib/sqlalchemy/dialects/mssql/base.py | 3 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_insert.py | 50 |
3 files changed, 61 insertions, 1 deletions
diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst index 6b842ab07..8ac241889 100644 --- a/doc/build/changelog/changelog_11.rst +++ b/doc/build/changelog/changelog_11.rst @@ -27,6 +27,15 @@ Fixed bug where an INSERT from SELECT where the source table contains an autoincrementing Sequence would fail to compile correctly. + .. change:: 3876 + :tags: bug, mssql + :tickets: 3876 + + Fixed bug where SQL Server dialects would attempt to select the + last row identity for an INSERT from SELECT, failing in the case when + the SELECT has no rows. For such a statement, + the inline flag is set to True indicating no last primary key + should be fetched. .. change:: 3875 :tags: bug, oracle diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py index 19558a2d6..1a1f7f06d 100644 --- a/lib/sqlalchemy/dialects/mssql/base.py +++ b/lib/sqlalchemy/dialects/mssql/base.py @@ -1022,7 +1022,8 @@ class MSExecutionContext(default.DefaultExecutionContext): else: self._enable_identity_insert = False - self._select_lastrowid = insert_has_sequence and \ + self._select_lastrowid = not self.compiled.inline and \ + insert_has_sequence and \ not self.compiled.returning and \ not self._enable_identity_insert and \ not self.executemany diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py index 70e8a6b17..c0b6b18eb 100644 --- a/lib/sqlalchemy/testing/suite/test_insert.py +++ b/lib/sqlalchemy/testing/suite/test_insert.py @@ -141,6 +141,56 @@ class InsertBehaviorTest(fixtures.TablesTest): assert len(r.fetchall()) @requirements.insert_from_select + def test_insert_from_select_autoinc(self): + src_table = self.tables.manual_pk + dest_table = self.tables.autoinc_pk + config.db.execute( + src_table.insert(), + [ + dict(id=1, data="data1"), + dict(id=2, data="data2"), + dict(id=3, data="data3"), + ] + ) + + result = config.db.execute( + dest_table.insert(). + from_select( + ("data",), + select([src_table.c.data]). + where(src_table.c.data.in_(["data2", "data3"])) + ) + ) + + eq_(result.inserted_primary_key, [None]) + + result = config.db.execute( + select([dest_table.c.data]).order_by(dest_table.c.data) + ) + eq_(result.fetchall(), [("data2", ), ("data3", )]) + + @requirements.insert_from_select + def test_insert_from_select_autoinc_no_rows(self): + src_table = self.tables.manual_pk + dest_table = self.tables.autoinc_pk + + result = config.db.execute( + dest_table.insert(). + from_select( + ("data",), + select([src_table.c.data]). + where(src_table.c.data.in_(["data2", "data3"])) + ) + ) + eq_(result.inserted_primary_key, [None]) + + result = config.db.execute( + select([dest_table.c.data]).order_by(dest_table.c.data) + ) + + eq_(result.fetchall(), []) + + @requirements.insert_from_select def test_insert_from_select(self): table = self.tables.manual_pk config.db.execute( |