--- /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.
stmt,
adapter.traverse(c),
{"result_map_targets": (c,)},
+ fallback_label_name=c._non_anon_label,
)
for c in expression._select_iterables(returning_cols)
]
from ... import types as sqltypes
from ... import util
from ...engine import cursor as _cursor
+from ...sql import expression
from ...util import compat
self.cursor,
[
(getattr(col, "name", col._anon_name_label), None)
- for col in self.compiled.returning
+ for col in expression._select_iterables(
+ self.compiled.returning
+ )
],
initial_buffer=[tuple(returning_params)],
)
-
self.cursor_fetch_strategy = fetch_strategy
def create_cursor(self):
def returning_clause(self, stmt, returning_cols):
columns = [
- self._label_returning_column(stmt, c)
+ self._label_returning_column(
+ stmt, c, fallback_label_name=c._non_anon_label
+ )
for c in expression._select_iterables(returning_cols)
]
)
self._result_columns.append((keyname, name, objects, type_))
- def _label_returning_column(self, stmt, column, column_clause_args=None):
+ def _label_returning_column(
+ self, stmt, column, column_clause_args=None, **kw
+ ):
"""Render a column with necessary labels inside of a RETURNING clause.
This method is provided for individual dialects in place of calling
True,
False,
{} if column_clause_args is None else column_clause_args,
+ **kw
)
def _label_select_column(
"_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(
def returning_clause(self, stmt, returning_cols):
columns = [
- self._label_select_column(None, c, True, False, {})
+ self._label_select_column(
+ None, c, True, False, {}, fallback_label_name=c._non_anon_label
+ )
for c in base._select_iterables(returning_cols)
]
self = self.set_label_style(LABEL_STYLE_DISAMBIGUATE_ONLY)
return self
- def _generate_columns_plus_names(self, anon_for_dupe_key):
+ def _generate_columns_plus_names(self, anon_for_dupe_key, cols=None):
"""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 AssertsCompiledSQL
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.assertions import is_
+from sqlalchemy.types import TypeEngine
from sqlalchemy.util import OrderedDict
from sqlalchemy.util import u
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(
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
"""
- __dialect__ = "default"
+ __dialect__ = "default_enhanced"
table1 = table("some_table", column("name"), column("value"))
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
+ )
+
+ # 1.4 behavior only; limited support for labels in RETURNING
+ 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",
+ )
+
def test_column_auto_label_dupes_label_style_none(self):
expr = self._fixture()
table1 = self.table1
# not sure if this SQL is right but this is what it was
# before the new labeling, just different label name
"SELECT value = 0 AS value, value",
+ dialect="default",
)
def test_label_auto_label_use_labels(self):
"inserted_primary_key",
)
+ @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", "firebird")
def test_literal_returning(self, connection):
if testing.against("postgresql"):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
