PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-05-14 20:57:19 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge "ensure _ORMJoin transfers parententity from left side" into rel_1_4

Browse Source
This commit is contained in:
mike bayer
2022-10-29 01:35:08 +00:00
committed by Gerrit Code Review
parent 2d7582e4ea 92468b1806
commit f030785d54
5 changed files with 302 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/build/changelog/unreleased_14/8721.rst
Vendored
+17
View File
@@ -0,0 +1,17 @@
.. change::
:tags: bug, orm
:tickets: 8721
Fixed bug involving :class:`.Select` constructs which used a combination of
:meth:`.Select.select_from` with an ORM entity followed by
:meth:`.Select.join` against the entity sent in
:meth:`.Select.select_from`, as well as using plain
:meth:`.Select.join_from`, which when combined with a columns clause that
didn't explicitly include that entity would then cause "automatic WHERE
criteria" features such as the IN expression required for a single-table
inheritance subclass, as well as the criteria set up by the
:func:`_orm.with_loader_criteria` option, to not be rendered for that
entity. The correct entity is now transferred to the :class:`.Join` object
that's generated internally, so that the criteria against the left
side entity is correctly added.
lib/sqlalchemy/orm/context.py
+1
View File
@@ -2253,6 +2253,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
for fromclause in self.from_clauses:
ext_info = fromclause._annotations.get("parententity", None)
if (
ext_info
and (
lib/sqlalchemy/orm/util.py
+26 -5
View File
@@ -1774,6 +1774,24 @@ class _ORMJoin(expression.Join):
self._target_adapter = target_adapter
# we don't use the normal coercions logic for _ORMJoin
# (probably should), so do some gymnastics to get the entity.
# logic here is for #8721, which was a major bug in 1.4
# for almost two years, not reported/fixed until 1.4.43 (!)
if left_info.is_selectable:
parententity = left_selectable._annotations.get(
"parententity", None
)
elif left_info.is_mapper or left_info.is_aliased_class:
parententity = left_info
else:
parententity = None
if parententity is not None:
self._annotations = self._annotations.union(
{"parententity": parententity}
)
augment_onclause = onclause is None and _extra_criteria
expression.Join.__init__(self, left, right, onclause, isouter, full)
@@ -1875,13 +1893,16 @@ def join(
join(User.addresses).\
filter(Address.email_address=='foo@bar.com')
See :ref:`orm_queryguide_joins` for information on modern usage
of ORM level joins.
.. warning:: using :func:`_orm.join` directly may not work properly
with modern ORM options such as :func:`_orm.with_loader_criteria`.
It is strongly recommended to use the idiomatic join patterns
provided by methods such as :meth:`.Select.join` and
:meth:`.Select.join_from` when creating ORM joins.
.. deprecated:: 0.8
.. seealso::
the ``join_to_left`` parameter is deprecated, and will be removed
in a future release. The parameter has no effect.
:ref:`orm_queryguide_joins` - in the :ref:`queryguide_toplevel` for
background on idiomatic ORM join patterns
"""
return _ORMJoin(left, right, onclause, isouter, full)
test/orm/inheritance/test_single.py
+119
View File
@@ -14,6 +14,7 @@ from sqlalchemy import util
from sqlalchemy.orm import aliased
from sqlalchemy.orm import Bundle
from sqlalchemy.orm import column_property
from sqlalchemy.orm import join as orm_join
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
@@ -395,6 +396,124 @@ class SingleInheritanceTest(testing.AssertsCompiledSQL, fixtures.MappedTest):
"WHERE employees_1.type IN (__[POSTCOMPILE_type_1])",
)
@testing.combinations(
(
lambda Engineer, Report: select(Report)
.select_from(Engineer)
.join(Engineer.reports),
),
(
lambda Engineer, Report: select(Report).select_from(
orm_join(Engineer, Report, Engineer.reports)
),
),
(
lambda Engineer, Report: select(Report).join_from(
Engineer, Report, Engineer.reports
),
),
argnames="stmt_fn",
)
@testing.combinations(True, False, argnames="alias_engineer")
def test_select_from_w_join_left(self, stmt_fn, alias_engineer):
"""test #8721"""
Engineer = self.classes.Engineer
Report = self.classes.Report
if alias_engineer:
Engineer = aliased(Engineer)
stmt = testing.resolve_lambda(
stmt_fn, Engineer=Engineer, Report=Report
)
if alias_engineer:
self.assert_compile(
stmt,
"SELECT reports.report_id, reports.employee_id, reports.name "
"FROM employees AS employees_1 JOIN reports "
"ON employees_1.employee_id = reports.employee_id "
"WHERE employees_1.type IN (__[POSTCOMPILE_type_1])",
)
else:
self.assert_compile(
stmt,
"SELECT reports.report_id, reports.employee_id, reports.name "
"FROM employees JOIN reports ON employees.employee_id = "
"reports.employee_id "
"WHERE employees.type IN (__[POSTCOMPILE_type_1])",
)
@testing.combinations(
(
lambda Engineer, Report: select(
Report.report_id, Engineer.employee_id
)
.select_from(Engineer)
.join(Engineer.reports),
),
(
lambda Engineer, Report: select(
Report.report_id, Engineer.employee_id
).select_from(orm_join(Engineer, Report, Engineer.reports)),
),
(
lambda Engineer, Report: select(
Report.report_id, Engineer.employee_id
).join_from(Engineer, Report, Engineer.reports),
),
)
def test_select_from_w_join_left_including_entity(self, stmt_fn):
"""test #8721"""
Engineer = self.classes.Engineer
Report = self.classes.Report
stmt = testing.resolve_lambda(
stmt_fn, Engineer=Engineer, Report=Report
)
self.assert_compile(
stmt,
"SELECT reports.report_id, employees.employee_id "
"FROM employees JOIN reports ON employees.employee_id = "
"reports.employee_id "
"WHERE employees.type IN (__[POSTCOMPILE_type_1])",
)
@testing.combinations(
(
lambda Engineer, Report: select(Report).join(
Report.employee.of_type(Engineer)
),
),
(
lambda Engineer, Report: select(Report).select_from(
orm_join(Report, Engineer, Report.employee.of_type(Engineer))
)
),
(
lambda Engineer, Report: select(Report).join_from(
Report, Engineer, Report.employee.of_type(Engineer)
),
),
)
def test_select_from_w_join_right(self, stmt_fn):
"""test #8721"""
Engineer = self.classes.Engineer
Report = self.classes.Report
stmt = testing.resolve_lambda(
stmt_fn, Engineer=Engineer, Report=Report
)
self.assert_compile(
stmt,
"SELECT reports.report_id, reports.employee_id, reports.name "
"FROM reports JOIN employees ON employees.employee_id = "
"reports.employee_id AND employees.type "
"IN (__[POSTCOMPILE_type_1])",
)
def test_from_statement_select(self):
Engineer = self.classes.Engineer
test/orm/test_relationship_criteria.py
+139
View File
@@ -16,6 +16,7 @@ from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.orm import aliased
from sqlalchemy.orm import defer
from sqlalchemy.orm import join as orm_join
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import lazyload
from sqlalchemy.orm import registry
@@ -264,6 +265,144 @@ class LoaderCriteriaTest(_Fixtures, testing.AssertsCompiledSQL):
"WHERE users.name != :name_1",
)
@testing.combinations(
(
lambda User, Address: select(Address)
.select_from(User)
.join(User.addresses)
.options(with_loader_criteria(User, User.name != "name")),
),
(
lambda User, Address: select(Address)
.select_from(orm_join(User, Address, User.addresses))
.options(with_loader_criteria(User, User.name != "name")),
),
(
lambda User, Address: select(Address)
.join_from(User, Address, User.addresses)
.options(with_loader_criteria(User, User.name != "name")),
),
argnames="stmt_fn",
)
@testing.combinations(True, False, argnames="alias_user")
def test_criteria_select_from_w_join_left(
self, user_address_fixture, stmt_fn, alias_user
):
"""test #8721"""
User, Address = user_address_fixture
if alias_user:
User = aliased(User)
stmt = testing.resolve_lambda(stmt_fn, User=User, Address=Address)
if alias_user:
self.assert_compile(
stmt,
"SELECT addresses.id, addresses.user_id, "
"addresses.email_address FROM users AS users_1 "
"JOIN addresses ON users_1.id = addresses.user_id "
"WHERE users_1.name != :name_1",
)
else:
self.assert_compile(
stmt,
"SELECT addresses.id, addresses.user_id, "
"addresses.email_address "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"WHERE users.name != :name_1",
)
@testing.combinations(
(
lambda User, Address: select(Address.id, User.id)
.select_from(User)
.join(User.addresses)
.options(with_loader_criteria(User, User.name != "name")),
),
(
lambda User, Address: select(Address.id, User.id)
.select_from(orm_join(User, Address, User.addresses))
.options(with_loader_criteria(User, User.name != "name")),
),
(
lambda User, Address: select(Address.id, User.id)
.join_from(User, Address, User.addresses)
.options(with_loader_criteria(User, User.name != "name")),
),
argnames="stmt_fn",
)
@testing.combinations(True, False, argnames="alias_user")
def test_criteria_select_from_w_join_left_including_entity(
self, user_address_fixture, stmt_fn, alias_user
):
"""test #8721"""
User, Address = user_address_fixture
if alias_user:
User = aliased(User)
stmt = testing.resolve_lambda(stmt_fn, User=User, Address=Address)
if alias_user:
self.assert_compile(
stmt,
"SELECT addresses.id, users_1.id AS id_1 "
"FROM users AS users_1 JOIN addresses "
"ON users_1.id = addresses.user_id "
"WHERE users_1.name != :name_1",
)
else:
self.assert_compile(
stmt,
"SELECT addresses.id, users.id AS id_1 "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"WHERE users.name != :name_1",
)
@testing.combinations(
(
lambda User, Address: select(Address)
.select_from(User)
.join(User.addresses)
.options(
with_loader_criteria(Address, Address.email_address != "email")
),
),
(
# for orm_join(), this is set up before we have the context
# available that allows with_loader_criteria to be set up
# correctly
lambda User, Address: select(Address)
.select_from(orm_join(User, Address, User.addresses))
.options(
with_loader_criteria(Address, Address.email_address != "email")
),
testing.fails("not implemented right now"),
),
(
lambda User, Address: select(Address)
.join_from(User, Address, User.addresses)
.options(
with_loader_criteria(Address, Address.email_address != "email")
),
),
argnames="stmt_fn",
)
def test_criteria_select_from_w_join_right(
self, user_address_fixture, stmt_fn
):
"""test #8721"""
User, Address = user_address_fixture
stmt = testing.resolve_lambda(stmt_fn, User=User, Address=Address)
self.assert_compile(
stmt,
"SELECT addresses.id, addresses.user_id, addresses.email_address "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"AND addresses.email_address != :email_address_1",
)
@testing.combinations(
"select",
"joined",