diff options
| -rw-r--r-- | doc/build/changelog/unreleased_13/5813.rst | 9 | ||||
| -rw-r--r-- | lib/sqlalchemy/dialects/oracle/base.py | 1 | ||||
| -rw-r--r-- | test/sql/test_returning.py | 84 |
3 files changed, 94 insertions, 0 deletions
diff --git a/doc/build/changelog/unreleased_13/5813.rst b/doc/build/changelog/unreleased_13/5813.rst new file mode 100644 index 000000000..d6483a26f --- /dev/null +++ b/doc/build/changelog/unreleased_13/5813.rst @@ -0,0 +1,9 @@ +.. change:: + :tags: bug, oracle + :tickets: 5813 + + Fixed regression in Oracle dialect introduced by :ticket:`4894` in + SQLAlchemy 1.3.11 where use of a SQL expression in RETURNING for an UPDATE + would fail to compile, due to a check for "server_default" when an + arbitrary SQL expression is not a column. + diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py index 2b4b268c2..1cc8b7aef 100644 --- a/lib/sqlalchemy/dialects/oracle/base.py +++ b/lib/sqlalchemy/dialects/oracle/base.py @@ -1017,6 +1017,7 @@ class OracleCompiler(compiler.SQLCompiler): ): if ( self.isupdate + and isinstance(column, sa_schema.Column) and isinstance(column.server_default, Computed) and not self.dialect._supports_update_returning_computed_cols ): diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py index 9f2afd7b7..187cf0dd0 100644 --- a/test/sql/test_returning.py +++ b/test/sql/test_returning.py @@ -11,6 +11,7 @@ from sqlalchemy import select from sqlalchemy import Sequence from sqlalchemy import String from sqlalchemy import testing +from sqlalchemy import type_coerce from sqlalchemy import update from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL @@ -121,6 +122,7 @@ class ReturningTest(fixtures.TablesTest, AssertsExecutionResults): Column("persons", Integer), Column("full", Boolean), Column("goofy", GoofyType(50)), + Column("strval", String(50)), ) def test_column_targeting(self, connection): @@ -197,6 +199,88 @@ class ReturningTest(fixtures.TablesTest, AssertsExecutionResults): ) eq_(result2.fetchall(), [(1, True), (2, False)]) + @testing.fails_on( + "mssql", + "driver has unknown issue with string concatenation " + "in INSERT RETURNING", + ) + def test_insert_returning_w_expression_one(self, connection): + table = self.tables.tables + result = connection.execute( + table.insert().returning(table.c.strval + "hi"), + {"persons": 5, "full": False, "strval": "str1"}, + ) + + eq_(result.fetchall(), [("str1hi",)]) + + result2 = connection.execute( + select(table.c.id, table.c.strval).order_by(table.c.id) + ) + eq_(result2.fetchall(), [(1, "str1")]) + + def test_insert_returning_w_type_coerce_expression(self, connection): + table = self.tables.tables + result = connection.execute( + table.insert().returning(type_coerce(table.c.goofy, String)), + {"persons": 5, "goofy": "somegoofy"}, + ) + + eq_(result.fetchall(), [("FOOsomegoofy",)]) + + result2 = connection.execute( + select(table.c.id, table.c.goofy).order_by(table.c.id) + ) + eq_(result2.fetchall(), [(1, "FOOsomegoofyBAR")]) + + def test_update_returning_w_expression_one(self, connection): + table = self.tables.tables + connection.execute( + table.insert(), + [ + {"persons": 5, "full": False, "strval": "str1"}, + {"persons": 3, "full": False, "strval": "str2"}, + ], + ) + + result = connection.execute( + table.update() + .where(table.c.persons > 4) + .values(full=True) + .returning(table.c.strval + "hi") + ) + eq_(result.fetchall(), [("str1hi",)]) + + result2 = connection.execute( + select(table.c.id, table.c.strval).order_by(table.c.id) + ) + eq_(result2.fetchall(), [(1, "str1"), (2, "str2")]) + + def test_update_returning_w_type_coerce_expression(self, connection): + table = self.tables.tables + connection.execute( + table.insert(), + [ + {"persons": 5, "goofy": "somegoofy1"}, + {"persons": 3, "goofy": "somegoofy2"}, + ], + ) + + result = connection.execute( + table.update() + .where(table.c.persons > 4) + .values(goofy="newgoofy") + .returning(type_coerce(table.c.goofy, String)) + ) + eq_(result.fetchall(), [("FOOnewgoofy",)]) + + result2 = connection.execute( + select(table.c.id, table.c.goofy).order_by(table.c.id) + ) + eq_( + result2.fetchall(), + [(1, "FOOnewgoofyBAR"), (2, "FOOsomegoofy2BAR")], + ) + @testing.requires.full_returning def test_update_full_returning(self, connection): table = self.tables.tables |