From: Mike Bayer Date: Fri, 8 Aug 2025 16:17:16 +0000 (-0400) Subject: warn on failed aliased X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=0439f17fd5a2fec48e0c3f4f4e5d397c6eff1edf;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git warn on failed aliased The :func:`_orm.aliased` object now emits warnings when an attribute is accessed on an aliased class that cannot be located in the target selectable, for those cases where the :func:`_orm.aliased` is against a different FROM clause than the regular mapped table (such as a subquery). This helps users identify cases where column names don't match between the aliased class and the underlying selectable. When :paramref:`_orm.aliased.adapt_on_names` is ``True``, the warning suggests checking the column name; when ``False``, it suggests using the ``adapt_on_names`` parameter for name-based matching. Fixes: #12838 Change-Id: I4294b57f24dd3fd7741e0bcbd9b521c841ace903 --- diff --git a/doc/build/changelog/unreleased_21/12838.rst b/doc/build/changelog/unreleased_21/12838.rst new file mode 100644 index 0000000000..2dd4a77b85 --- /dev/null +++ b/doc/build/changelog/unreleased_21/12838.rst @@ -0,0 +1,13 @@ +.. change:: + :tags: usecase, orm + :tickets: 12838 + + The :func:`_orm.aliased` object now emits warnings when an attribute is + accessed on an aliased class that cannot be located in the target + selectable, for those cases where the :func:`_orm.aliased` is against a + different FROM clause than the regular mapped table (such as a subquery). + This helps users identify cases where column names don't match between the + aliased class and the underlying selectable. When + :paramref:`_orm.aliased.adapt_on_names` is ``True``, the warning suggests + checking the column name; when ``False``, it suggests using the + ``adapt_on_names`` parameter for name-based matching. diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py index eb8472993a..1184a223af 100644 --- a/lib/sqlalchemy/orm/util.py +++ b/lib/sqlalchemy/orm/util.py @@ -1187,14 +1187,27 @@ class AliasedInsp( if key: d["proxy_key"] = key - # IMO mypy should see this one also as returning the same type - # we put into it, but it's not - return ( - self._adapter.traverse(expr) - ._annotate(d) - ._set_propagate_attrs( - {"compile_state_plugin": "orm", "plugin_subject": self} - ) + # userspace adapt of an attribute from AliasedClass; validate that + # it actually was present + adapted = self._adapter.adapt_check_present(expr) + if adapted is None: + adapted = expr + if self._adapter.adapt_on_names: + util.warn_limited( + "Did not locate an expression in selectable for " + "attribute %r; ensure name is correct in expression", + (key,), + ) + else: + util.warn_limited( + "Did not locate an expression in selectable for " + "attribute %r; to match by name, use the " + "adapt_on_names parameter", + (key,), + ) + + return adapted._annotate(d)._set_propagate_attrs( + {"compile_state_plugin": "orm", "plugin_subject": self} ) if TYPE_CHECKING: diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py index 128fc6d345..f29e4a65a6 100644 --- a/lib/sqlalchemy/sql/util.py +++ b/lib/sqlalchemy/sql/util.py @@ -1348,9 +1348,7 @@ class ColumnAdapter(ClauseAdapter): adapt_clause = traverse adapt_list = ClauseAdapter.copy_and_process - def adapt_check_present( - self, col: ColumnElement[Any] - ) -> Optional[ColumnElement[Any]]: + def adapt_check_present(self, col: _ET) -> Optional[_ET]: newcol = self.columns[col] if newcol is col and self._corresponding_column(col, True) is None: diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py index ae0c147c71..c67269c6c7 100644 --- a/test/orm/test_froms.py +++ b/test/orm/test_froms.py @@ -2842,42 +2842,22 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL): self.assert_compile(q, exp) def test_aliased_adapt_on_names(self): - User, Address = self.classes.User, self.classes.Address - - sess = fixture_session() - agg_address = sess.query( + User, Address = self.classes("User", "Address") + agg_address = select( Address.id, func.sum(func.length(Address.email_address)).label( "email_address" ), ).group_by(Address.user_id) - - ag1 = aliased(Address, agg_address.subquery()) ag2 = aliased(Address, agg_address.subquery(), adapt_on_names=True) - # first, without adapt on names, 'email_address' isn't matched up - we - # get the raw "address" element in the SELECT - self.assert_compile( - sess.query(User, ag1.email_address) - .join(ag1, User.addresses) - .filter(ag1.email_address > 5), - "SELECT users.id " - "AS users_id, users.name AS users_name, addresses.email_address " - "AS addresses_email_address FROM users JOIN " - "(SELECT addresses.id AS id, sum(length(addresses.email_address)) " - "AS email_address FROM addresses GROUP BY addresses.user_id) AS " - "anon_1 ON users.id = addresses.user_id, addresses " - "WHERE addresses.email_address > :email_address_1", - ) - # second, 'email_address' matches up to the aggregate, and we get a # smooth JOIN from users->subquery and that's it self.assert_compile( - sess.query(User, ag2.email_address) + select(User, ag2.email_address) .join(ag2, User.addresses) .filter(ag2.email_address > 5), - "SELECT users.id AS users_id, users.name AS users_name, " - "anon_1.email_address AS anon_1_email_address FROM users " + "SELECT users.id, users.name, anon_1.email_address FROM users " "JOIN (" "SELECT addresses.id AS id, sum(length(addresses.email_address)) " "AS email_address FROM addresses GROUP BY addresses.user_id) AS " @@ -2885,6 +2865,68 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL): "WHERE anon_1.email_address > :email_address_1", ) + def test_aliased_warns_missing_column(self): + User, Address = self.classes("User", "Address") + agg_address = select( + Address.id, + func.sum(func.length(Address.email_address)).label( + "email_address" + ), + ).group_by(Address.user_id) + + ag1 = aliased(Address, agg_address.subquery()) + + # without adapt on names, 'email_address' isn't matched up - we + # get the raw "address" element in the SELECT + with testing.expect_warnings( + r"Did not locate an expression in selectable for attribute " + r"'email_address'; to match by name, use the " + r"adapt_on_names parameter" + ): + self.assert_compile( + select(User, ag1.email_address) + .join(ag1, User.addresses) + .filter(ag1.email_address > 5), + "SELECT users.id, users.name, addresses.email_address " + "FROM users JOIN " + "(SELECT addresses.id AS id, " + "sum(length(addresses.email_address)) " + "AS email_address FROM addresses " + "GROUP BY addresses.user_id) AS " + "anon_1 ON users.id = addresses.user_id, addresses " + "WHERE addresses.email_address > :email_address_1", + ) + + def test_aliased_warns_unmatched_name(self): + User, Address = self.classes("User", "Address") + agg_address = select( + Address.id, + func.sum(func.length(Address.email_address)).label( + "email_address_misspelled" + ), + ).group_by(Address.user_id) + + ag1 = aliased(Address, agg_address.subquery(), adapt_on_names=True) + + # adapt_on_names is set but still wrong name + with testing.expect_warnings( + r"Did not locate an expression in selectable for attribute " + r"'email_address'; ensure name is correct in expression" + ): + self.assert_compile( + select(User, ag1.email_address) + .join(ag1, User.addresses) + .filter(ag1.email_address > 5), + "SELECT users.id, users.name, addresses.email_address " + "FROM users JOIN " + "(SELECT addresses.id AS id, " + "sum(length(addresses.email_address)) " + "AS email_address_misspelled FROM addresses " + "GROUP BY addresses.user_id) AS " + "anon_1 ON users.id = addresses.user_id, addresses " + "WHERE addresses.email_address > :email_address_1", + ) + class SelectFromTest(QueryTest, AssertsCompiledSQL): run_setup_mappers = None