--- /dev/null
+.. change::
+ :tags: bug, postgresql, mssql
+ :tickets: 8770
+
+ For the PostgreSQL and SQL Server dialects only, adjusted the compiler so
+ that when rendering column expressions in the RETURNING clause, the "non
+ anon" label that's used in SELECT statements is suggested for SQL
+ expression elements that generate a label; the primary example is a SQL
+ function that may be emitting as part of the column's type, where the label
+ name should match the column's name by default. This restores a not-well
+ defined behavior that had changed in version 1.4.21 due to :ticket:`6718`,
+ :ticket:`6710`. The Oracle dialect has a different RETURNING implementation
+ and was not affected by this issue. Version 2.0 features an across the
+ board change for its widely expanded support of RETURNING on other
+ backends.
+
+
+.. change::
+ :tags: bug, oracle
+
+ Fixed issue in the Oracle dialect where an INSERT statement that used
+ ``insert(some_table).values(...).returning(some_table)`` against a full
+ :class:`.Table` object at once would fail to execute, raising an exception.
--- /dev/null
+.. change::
+ :tags: bug, sql
+ :tickets: 8770
+
+ The RETURNING clause now renders columns using the routine as that of the
+ :class:`.Select` to generate labels, which will include disambiguating
+ labels, as well as that a SQL function surrounding a named column will be
+ labeled using the column name itself. This is a more comprehensive change
+ than a similar one made for the 1.4 series that adjusted the function label
+ issue only.
columns = [
self._label_returning_column(
stmt,
- adapter.traverse(c),
+ adapter.traverse(column),
populate_result_map,
- {"result_map_targets": (c,)},
+ {"result_map_targets": (column,)},
+ fallback_label_name=fallback_label_name,
+ column_is_repeated=repeated,
+ name=name,
+ proxy_name=proxy_name,
+ **kw,
+ )
+ for (
+ name,
+ proxy_name,
+ fallback_label_name,
+ column,
+ repeated,
+ ) in stmt._generate_columns_plus_names(
+ True, cols=expression._select_iterables(returning_cols)
)
- for c in expression._select_iterables(returning_cols)
]
return "OUTPUT " + ", ".join(columns)
"_label_select_column is only relevant within "
"the columns clause of a SELECT or RETURNING"
)
-
if isinstance(column, elements.Label):
if col_expr is not column:
result_expr = _CompileLabel(
populate_result_map: bool,
**kw: Any,
) -> str:
+
columns = [
- self._label_returning_column(stmt, c, populate_result_map, **kw)
- for c in base._select_iterables(returning_cols)
+ self._label_returning_column(
+ stmt,
+ column,
+ populate_result_map,
+ fallback_label_name=fallback_label_name,
+ column_is_repeated=repeated,
+ name=name,
+ proxy_name=proxy_name,
+ **kw,
+ )
+ for (
+ name,
+ proxy_name,
+ fallback_label_name,
+ column,
+ repeated,
+ ) in stmt._generate_columns_plus_names(
+ True, cols=base._select_iterables(returning_cols)
+ )
]
return "RETURNING " + ", ".join(columns)
from .selectable import HasCTE
from .selectable import HasPrefixes
from .selectable import Join
+from .selectable import SelectLabelStyle
from .selectable import TableClause
from .selectable import TypedReturnsRows
from .sqltypes import NullType
] = util.EMPTY_DICT
named_with_column = False
+ _label_style: SelectLabelStyle = (
+ SelectLabelStyle.LABEL_STYLE_DISAMBIGUATE_ONLY
+ )
table: _DMLTableElement
_return_defaults = False
_label_style: SelectLabelStyle = LABEL_STYLE_NONE
def _generate_columns_plus_names(
- self, anon_for_dupe_key: bool
+ self,
+ anon_for_dupe_key: bool,
+ cols: Optional[_SelectIterable] = None,
) -> List[_ColumnsPlusNames]:
"""Generate column names as rendered in a SELECT statement by
the compiler.
_column_naming_convention as well.
"""
- cols = self._all_selected_columns
+
+ if cols is None:
+ cols = self._all_selected_columns
key_naming_convention = SelectState._column_naming_convention(
self._label_style
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing.assertions import eq_ignore_whitespace
+from sqlalchemy.types import TypeEngine
tbl = table("t", column("a"))
"Latin1_General_CS_AS_KS_WS_CI ASC",
)
+ @testing.fixture
+ def column_expression_fixture(self):
+ class MyString(TypeEngine):
+ def column_expression(self, column):
+ return func.lower(column)
+
+ return table(
+ "some_table", column("name", String), column("value", MyString)
+ )
+
+ @testing.combinations("columns", "table", argnames="use_columns")
+ def test_plain_returning_column_expression(
+ self, column_expression_fixture, use_columns
+ ):
+ """test #8770"""
+ table1 = column_expression_fixture
+
+ if use_columns == "columns":
+ stmt = insert(table1).returning(table1)
+ else:
+ stmt = insert(table1).returning(table1.c.name, table1.c.value)
+
+ self.assert_compile(
+ stmt,
+ "INSERT INTO some_table (name, value) OUTPUT inserted.name, "
+ "lower(inserted.value) AS value VALUES (:name, :value)",
+ )
+
def test_join_with_hint(self):
t1 = table(
"t1",
from sqlalchemy import func
from sqlalchemy import Identity
from sqlalchemy import Index
+from sqlalchemy import insert
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import literal_column
from sqlalchemy.testing.assertions import eq_ignore_whitespace
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
+from sqlalchemy.types import TypeEngine
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"t1.c2, t1.c3 INTO :ret_0, :ret_1",
)
+ @testing.fixture
+ def column_expression_fixture(self):
+ class MyString(TypeEngine):
+ def column_expression(self, column):
+ return func.lower(column)
+
+ return table(
+ "some_table", column("name", String), column("value", MyString)
+ )
+
+ @testing.combinations("columns", "table", argnames="use_columns")
+ def test_plain_returning_column_expression(
+ self, column_expression_fixture, use_columns
+ ):
+ """test #8770"""
+ table1 = column_expression_fixture
+
+ if use_columns == "columns":
+ stmt = insert(table1).returning(table1)
+ else:
+ stmt = insert(table1).returning(table1.c.name, table1.c.value)
+
+ self.assert_compile(
+ stmt,
+ "INSERT INTO some_table (name, value) VALUES (:name, :value) "
+ "RETURNING some_table.name, lower(some_table.value) "
+ "INTO :ret_0, :ret_1",
+ )
+
def test_returning_insert_computed(self):
m = MetaData()
t1 = Table(
from sqlalchemy.testing.assertions import eq_ignore_whitespace
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.assertions import is_
+from sqlalchemy.types import TypeEngine
from sqlalchemy.util import OrderedDict
dialect=dialect,
)
+ @testing.fixture
+ def column_expression_fixture(self):
+ class MyString(TypeEngine):
+ def column_expression(self, column):
+ return func.lower(column)
+
+ return table(
+ "some_table", column("name", String), column("value", MyString)
+ )
+
+ @testing.combinations("columns", "table", argnames="use_columns")
+ def test_plain_returning_column_expression(
+ self, column_expression_fixture, use_columns
+ ):
+ """test #8770"""
+ table1 = column_expression_fixture
+
+ if use_columns == "columns":
+ stmt = insert(table1).returning(table1)
+ else:
+ stmt = insert(table1).returning(table1.c.name, table1.c.value)
+
+ self.assert_compile(
+ stmt,
+ "INSERT INTO some_table (name, value) "
+ "VALUES (%(name)s, %(value)s) RETURNING some_table.name, "
+ "lower(some_table.value) AS value",
+ )
+
def test_create_drop_enum(self):
# test escaping and unicode within CREATE TYPE for ENUM
typ = postgresql.ENUM("val1", "val2", "val's 3", "méil", name="myname")
from sqlalchemy import Boolean
from sqlalchemy import cast
from sqlalchemy import exc as exceptions
+from sqlalchemy import func
+from sqlalchemy import insert
from sqlalchemy import Integer
from sqlalchemy import literal_column
from sqlalchemy import MetaData
from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
+from sqlalchemy.types import TypeEngine
IDENT_LENGTH = 29
return SomeColThing
+ @testing.fixture
+ def compiler_column_fixture(self):
+ return self._fixture()
+
+ @testing.fixture
+ def column_expression_fixture(self):
+ class MyString(TypeEngine):
+ def column_expression(self, column):
+ return func.lower(column)
+
+ return table(
+ "some_table", column("name", String), column("value", MyString)
+ )
+
+ def test_plain_select_compiler_expression(self, compiler_column_fixture):
+ expr = compiler_column_fixture
+ table1 = self.table1
+
+ self.assert_compile(
+ select(
+ table1.c.name,
+ expr(table1.c.value),
+ ),
+ "SELECT some_table.name, SOME_COL_THING(some_table.value) "
+ "AS value FROM some_table",
+ )
+
+ def test_plain_select_column_expression(self, column_expression_fixture):
+ table1 = column_expression_fixture
+
+ self.assert_compile(
+ select(table1),
+ "SELECT some_table.name, lower(some_table.value) AS value "
+ "FROM some_table",
+ )
+
+ def test_plain_returning_compiler_expression(
+ self, compiler_column_fixture
+ ):
+ expr = compiler_column_fixture
+ table1 = self.table1
+
+ self.assert_compile(
+ insert(table1).returning(
+ table1.c.name,
+ expr(table1.c.value),
+ ),
+ "INSERT INTO some_table (name, value) VALUES (:name, :value) "
+ "RETURNING some_table.name, "
+ "SOME_COL_THING(some_table.value) AS value",
+ )
+
+ @testing.combinations("columns", "table", argnames="use_columns")
+ def test_plain_returning_column_expression(
+ self, column_expression_fixture, use_columns
+ ):
+ table1 = column_expression_fixture
+
+ if use_columns == "columns":
+ stmt = insert(table1).returning(table1)
+ else:
+ stmt = insert(table1).returning(table1.c.name, table1.c.value)
+
+ self.assert_compile(
+ stmt,
+ "INSERT INTO some_table (name, value) VALUES (:name, :value) "
+ "RETURNING some_table.name, lower(some_table.value) AS value",
+ )
+
+ def test_select_dupes_column_expression(self, column_expression_fixture):
+ table1 = column_expression_fixture
+
+ self.assert_compile(
+ select(table1.c.name, table1.c.value, table1.c.value),
+ "SELECT some_table.name, lower(some_table.value) AS value, "
+ "lower(some_table.value) AS value__1 FROM some_table",
+ )
+
+ def test_returning_dupes_column_expression(
+ self, column_expression_fixture
+ ):
+ table1 = column_expression_fixture
+
+ stmt = insert(table1).returning(
+ table1.c.name, table1.c.value, table1.c.value
+ )
+
+ self.assert_compile(
+ stmt,
+ "INSERT INTO some_table (name, value) VALUES (:name, :value) "
+ "RETURNING some_table.name, lower(some_table.value) AS value, "
+ "lower(some_table.value) AS value__1",
+ )
+
def test_column_auto_label_dupes_label_style_none(self):
expr = self._fixture()
table1 = self.table1
result = connection.execute(ins)
eq_(result.fetchall(), [(1, 1), (2, 2), (3, 3)])
+ @testing.fixture
+ def column_expression_fixture(self, metadata, connection):
+ class MyString(TypeDecorator):
+ cache_ok = True
+ impl = String(50)
+
+ def column_expression(self, column):
+ return func.lower(column)
+
+ t1 = Table(
+ "some_table",
+ metadata,
+ Column("name", String(50)),
+ Column("value", MyString(50)),
+ )
+ metadata.create_all(connection)
+ return t1
+
+ @testing.combinations("columns", "table", argnames="use_columns")
+ def test_plain_returning_column_expression(
+ self, column_expression_fixture, use_columns, connection
+ ):
+ """test #8770"""
+ table1 = column_expression_fixture
+
+ if use_columns == "columns":
+ stmt = (
+ insert(table1)
+ .values(name="n1", value="ValUE1")
+ .returning(table1)
+ )
+ else:
+ stmt = (
+ insert(table1)
+ .values(name="n1", value="ValUE1")
+ .returning(table1.c.name, table1.c.value)
+ )
+
+ result = connection.execute(stmt)
+ row = result.first()
+
+ eq_(row._mapping["name"], "n1")
+ eq_(row._mapping["value"], "value1")
+
@testing.fails_on_everything_except(
"postgresql", "mariadb>=10.5", "sqlite>=3.34"
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
