--- /dev/null
+.. change::
+ :tags: bug, orm, regression
+ :tickets: 9273
+
+ Fixed regression introduced in version 2.0.2 due to :ticket:`9217` where
+ using DML RETURNING statements, as well as
+ :meth:`_sql.Select.from_statement` constructs as was "fixed" in
+ :ticket:`9217`, in conjunction with ORM mapped classes that used
+ expressions such as with :func:`_orm.column_property`, would lead to an
+ internal error within Core where it would attempt to match the expression
+ by name. The fix repairs the Core issue, and also adjusts the fix in
+ :ticket:`9217` to not take effect for the DML RETURNING use case, where it
+ adds unnecessary overhead.
if orm_level_statement._returning:
fs = FromStatement(
- orm_level_statement._returning, dml_level_statement
+ orm_level_statement._returning,
+ dml_level_statement,
+ _adapt_on_names=False,
)
fs = fs.options(*orm_level_statement._with_options)
self.select_statement = fs
**kw: Any,
) -> ORMFromStatementCompileState:
+ assert isinstance(statement_container, FromStatement)
+
if compiler is not None:
toplevel = not compiler.stack
else:
# those columns completely, don't interfere with the compiler
# at all; just in ORM land, use an adapter to convert from
# our ORM columns to whatever columns are in the statement,
- # before we look in the result row. Always adapt on names
- # to accept cases such as issue #9217.
-
+ # before we look in the result row. Adapt on names
+ # to accept cases such as issue #9217, however also allow
+ # this to be overridden for cases such as #9273.
self._from_obj_alias = ORMStatementAdapter(
_TraceAdaptRole.ADAPT_FROM_STATEMENT,
self.statement,
- adapt_on_names=True,
+ adapt_on_names=statement_container._adapt_on_names,
)
return self
element: Union[ExecutableReturnsRows, TextClause]
+ _adapt_on_names: bool
+
_traverse_internals = [
("_raw_columns", InternalTraversal.dp_clauseelement_list),
("element", InternalTraversal.dp_clauseelement),
self,
entities: Iterable[_ColumnsClauseArgument[Any]],
element: Union[ExecutableReturnsRows, TextClause],
+ _adapt_on_names: bool = True,
):
self._raw_columns = [
coercions.expect(
self._label_style = (
element._label_style if is_select_base(element) else None
)
+ self._adapt_on_names = _adapt_on_names
def _compiler_dispatch(self, compiler, **kw):
from .elements import Grouping
from .elements import KeyedColumnElement
from .elements import Label
+from .elements import NamedColumn
from .elements import Null
from .elements import UnaryExpression
from .schema import Column
return "(%s)" % elements
def _get_batches(self, params: Iterable[Any]) -> Any:
-
lparams = list(params)
lenparams = len(lparams)
if lenparams > self.max_params:
def _corresponding_column(
self, col, require_embedded, _seen=util.EMPTY_SET
):
-
newcol = self.selectable.corresponding_column(
col, require_embedded=require_embedded
)
)
if newcol is not None:
return newcol
- if self.adapt_on_names and newcol is None:
+
+ if (
+ self.adapt_on_names
+ and newcol is None
+ and isinstance(col, NamedColumn)
+ ):
newcol = self.selectable.exported_columns.get(col.name)
return newcol
from sqlalchemy import Identity
from sqlalchemy import insert
from sqlalchemy import inspect
+from sqlalchemy import literal
from sqlalchemy import literal_column
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import update
from sqlalchemy.orm import aliased
+from sqlalchemy.orm import column_property
from sqlalchemy.orm import load_only
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.testing import mock
from sqlalchemy.testing import provision
from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.testing.fixtures import fixture_session
-class NoReturningTest(fixtures.TestBase):
+class InsertStmtTest(fixtures.TestBase):
def test_no_returning_error(self, decl_base):
class A(fixtures.ComparableEntity, decl_base):
__tablename__ = "a"
[("d3", 5), ("d4", 6)],
)
+ def test_insert_from_select_col_property(self, decl_base):
+ """test #9273"""
+
+ class User(ComparableEntity, decl_base):
+ __tablename__ = "users"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ name: Mapped[str] = mapped_column()
+ age: Mapped[int] = mapped_column()
+
+ is_adult: Mapped[bool] = column_property(age >= 18)
+
+ decl_base.metadata.create_all(testing.db)
+
+ stmt = select(
+ literal(1).label("id"),
+ literal("John").label("name"),
+ literal(30).label("age"),
+ )
+
+ insert_stmt = (
+ insert(User)
+ .from_select(["id", "name", "age"], stmt)
+ .returning(User)
+ )
+
+ s = fixture_session()
+ result = s.scalars(insert_stmt)
+
+ eq_(result.all(), [User(id=1, name="John", age=30)])
+
class BulkDMLReturningInhTest:
def test_insert_col_key_also_works_currently(self):
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
+from sqlalchemy import literal
from sqlalchemy import literal_column
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.context import ORMSelectCompileState
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
+from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
from test.orm import _fixtures
eq_(q.all(), expected)
def test_unrelated_column(self):
- """Test for 9217"""
+ """Test for #9217"""
User = self.classes.User
s = select(User).from_statement(q)
sess = fixture_session()
res = sess.scalars(s).one()
- in_("name", res.__dict__)
- eq_(res.name, "sandy")
+ eq_(res, User(name="sandy", id=7))
+
+ def test_unrelated_column_col_prop(self, decl_base):
+ """Test for #9217 combined with #9273"""
+
+ class User(ComparableEntity, decl_base):
+ __tablename__ = "some_user_table"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ name: Mapped[str] = mapped_column()
+ age: Mapped[int] = mapped_column()
+
+ is_adult: Mapped[bool] = column_property(age >= 18)
+
+ stmt = select(
+ literal(1).label("id"),
+ literal("John").label("name"),
+ literal(30).label("age"),
+ )
+
+ s = select(User).from_statement(stmt)
+ sess = fixture_session()
+ res = sess.scalars(s).one()
+
+ eq_(res, User(name="John", age=30, id=1))
def test_expression_selectable_matches_mzero(self):
User, Address = self.classes.User, self.classes.Address
)
def test_this_thing_using_setup_joins_three(self):
-
j = t1.join(t2, t1.c.col1 == t2.c.col2)
s1 = select(j)
)
def test_this_thing_using_setup_joins_four(self):
-
j = t1.join(t2, t1.c.col1 == t2.c.col2)
s1 = select(j)
# not covered by a1, rejected by a2
is_(a3.columns[c2a1], c2a1)
+ @testing.combinations(True, False, argnames="colpresent")
+ @testing.combinations(True, False, argnames="adapt_on_names")
+ @testing.combinations(True, False, argnames="use_label")
+ def test_adapt_binary_col(self, colpresent, use_label, adapt_on_names):
+ """test #9273"""
+
+ if use_label:
+ stmt = select(t1.c.col1, (t1.c.col2 > 18).label("foo"))
+ else:
+ stmt = select(t1.c.col1, (t1.c.col2 > 18))
+
+ sq = stmt.subquery()
+
+ if colpresent:
+ s2 = select(sq.c[0], sq.c[1])
+ else:
+ s2 = select(sq.c[0])
+
+ a1 = sql_util.ColumnAdapter(s2, adapt_on_names=adapt_on_names)
+
+ is_(a1.columns[stmt.selected_columns[0]], s2.selected_columns[0])
+
+ if colpresent:
+ is_(a1.columns[stmt.selected_columns[1]], s2.selected_columns[1])
+ else:
+ is_(
+ a1.columns[stmt.selected_columns[1]],
+ a1.columns[stmt.selected_columns[1]],
+ )
+
class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
)
def test_adapt_select_w_unlabeled_fn(self):
-
expr = func.count(t1.c.col1)
stmt = select(t1, expr)
assert s2.is_derived_from(s1)
def test_aliasedselect_to_aliasedselect_straight(self):
-
# original issue from ticket #904
s1 = select(t1).alias("foo")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
