diff options
| -rw-r--r-- | doc/build/changelog/unreleased_12/4454.rst | 12 | ||||
| -rw-r--r-- | lib/sqlalchemy/orm/mapper.py | 2 | ||||
| -rw-r--r-- | lib/sqlalchemy/orm/query.py | 2 | ||||
| -rw-r--r-- | test/orm/inheritance/test_single.py | 213 |
4 files changed, 227 insertions, 2 deletions
diff --git a/doc/build/changelog/unreleased_12/4454.rst b/doc/build/changelog/unreleased_12/4454.rst new file mode 100644 index 000000000..75f698e50 --- /dev/null +++ b/doc/build/changelog/unreleased_12/4454.rst @@ -0,0 +1,12 @@ +.. change:: + :tags: bug, orm + :tickets: 4454 + + Fixed issue where when using single-table inheritance in conjunction with a + joined inheritance hierarchy that uses "with polymorphic" loading, the + "single table criteria" for that single-table entity could get confused for + that of other entities from the same hierarchy used in the same query.The + adaption of the "single table criteria" is made more specific to the target + entity to avoid it accidentally getting adapted to other tables in the + query. + diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py index 56ad965de..a394ec06e 100644 --- a/lib/sqlalchemy/orm/mapper.py +++ b/lib/sqlalchemy/orm/mapper.py @@ -2103,7 +2103,7 @@ class Mapper(InspectionAttr): @_memoized_configured_property def _single_table_criterion(self): if self.single and self.inherits and self.polymorphic_on is not None: - return self.polymorphic_on.in_( + return self.polymorphic_on._annotate({"parentmapper": self}).in_( m.polymorphic_identity for m in self.self_and_descendants ) else: diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py index 7ddcf10b0..2bd79a2cd 100644 --- a/lib/sqlalchemy/orm/query.py +++ b/lib/sqlalchemy/orm/query.py @@ -364,7 +364,6 @@ class Query(object): or "_orm_adapt" in elem._annotations or "parententity" in elem._annotations ): - e = adapter(elem) if e is not None: return e @@ -3933,6 +3932,7 @@ class Query(object): if single_crit is not None: if adapter: single_crit = adapter.traverse(single_crit) + single_crit = self._adapt_clause(single_crit, False, False) context.whereclause = sql.and_( sql.True_._ifnone(context.whereclause), single_crit diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py index aae9312d4..1b28974b7 100644 --- a/test/orm/inheritance/test_single.py +++ b/test/orm/inheritance/test_single.py @@ -4,6 +4,8 @@ from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import literal from sqlalchemy import literal_column +from sqlalchemy import null +from sqlalchemy import select from sqlalchemy import String from sqlalchemy import testing from sqlalchemy.orm import aliased @@ -16,6 +18,7 @@ from sqlalchemy.orm import relationship from sqlalchemy.orm import Session from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import subqueryload +from sqlalchemy.orm import with_polymorphic from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures @@ -1436,6 +1439,216 @@ class SingleOnJoinedTest(fixtures.MappedTest): self.assert_sql_count(testing.db, go, 1) +class SingleFromPolySelectableTest( + fixtures.DeclarativeMappedTest, AssertsCompiledSQL +): + __dialect__ = "default" + + @classmethod + def setup_classes(cls, with_polymorphic=None, include_sub_defaults=False): + Base = cls.DeclarativeBasic + + class Employee(Base): + __tablename__ = "employee" + id = Column(Integer, primary_key=True) + name = Column(String(50)) + type = Column(String(50)) + + __mapper_args__ = { + "polymorphic_identity": "employee", + "polymorphic_on": type, + } + + class Engineer(Employee): + __tablename__ = "engineer" + id = Column(Integer, ForeignKey("employee.id"), primary_key=True) + engineer_info = Column(String(50)) + manager_id = Column(ForeignKey("manager.id")) + __mapper_args__ = {"polymorphic_identity": "engineer"} + + class Manager(Employee): + __tablename__ = "manager" + id = Column(Integer, ForeignKey("employee.id"), primary_key=True) + manager_data = Column(String(50)) + __mapper_args__ = {"polymorphic_identity": "manager"} + + class Boss(Manager): + __mapper_args__ = {"polymorphic_identity": "boss"} + + def _with_poly_fixture(self): + employee = self.classes.Employee.__table__ + engineer = self.classes.Engineer.__table__ + manager = self.classes.Manager.__table__ + + poly = ( + select( + [ + employee.c.id, + employee.c.type, + employee.c.name, + manager.c.manager_data, + null().label("engineer_info"), + null().label("manager_id"), + ] + ) + .select_from(employee.join(manager)) + .apply_labels() + .union_all( + select( + [ + employee.c.id, + employee.c.type, + employee.c.name, + null().label("manager_data"), + engineer.c.engineer_info, + engineer.c.manager_id, + ] + ) + .select_from(employee.join(engineer)) + .apply_labels() + ) + .alias() + ) + + return poly + + def test_wpoly_single_inh_subclass(self): + poly = with_polymorphic( + self.classes.Employee, + [self.classes.Boss, self.classes.Manager, self.classes.Engineer], + self._with_poly_fixture(), + ) + s = Session() + q = s.query(poly.Boss) + self.assert_compile( + q, + "SELECT " + "anon_1.employee_id AS anon_1_employee_id, " + "anon_1.employee_name AS anon_1_employee_name, " + "anon_1.employee_type AS anon_1_employee_type, " + "anon_1.manager_manager_data AS anon_1_manager_manager_data " + "FROM " + "(SELECT " + "employee.id AS employee_id, employee.type AS employee_type, " + "employee.name AS employee_name, " + "manager.manager_data AS manager_manager_data, " + "NULL AS engineer_info, NULL AS manager_id FROM employee " + "JOIN manager ON employee.id = manager.id " + "UNION ALL " + "SELECT employee.id AS employee_id, " + "employee.type AS employee_type, " + "employee.name AS employee_name, NULL AS manager_data, " + "engineer.engineer_info AS engineer_engineer_info, " + "engineer.manager_id AS engineer_manager_id " + "FROM employee JOIN engineer ON employee.id = engineer.id) " + "AS anon_1 " + "WHERE anon_1.employee_type IN (:type_1)", + ) + + def test_query_wpoly_single_inh_subclass(self): + Boss = self.classes.Boss + + poly = self._with_poly_fixture() + + s = Session() + q = s.query(Boss).with_polymorphic(Boss, poly) + self.assert_compile( + q, + "SELECT anon_1.employee_id AS anon_1_employee_id, " + "anon_1.employee_name AS anon_1_employee_name, " + "anon_1.employee_type AS anon_1_employee_type, " + "anon_1.manager_manager_data AS anon_1_manager_manager_data " + "FROM (SELECT employee.id AS employee_id, employee.type " + "AS employee_type, employee.name AS employee_name, " + "manager.manager_data AS manager_manager_data, " + "NULL AS engineer_info, NULL AS manager_id FROM employee " + "JOIN manager ON employee.id = manager.id " + "UNION ALL SELECT employee.id AS employee_id, " + "employee.type AS employee_type, employee.name AS employee_name, " + "NULL AS manager_data, " + "engineer.engineer_info AS engineer_engineer_info, " + "engineer.manager_id AS engineer_manager_id " + "FROM employee JOIN engineer ON employee.id = engineer.id) " + "AS anon_1 WHERE anon_1.employee_type IN (:type_1)", + ) + + def test_single_inh_subclass_join_joined_inh_subclass(self): + Boss, Engineer = self.classes("Boss", "Engineer") + s = Session() + + q = s.query(Boss).join(Engineer, Engineer.manager_id == Boss.id) + + self.assert_compile( + q, + "SELECT manager.id AS manager_id, employee.id AS employee_id, " + "employee.name AS employee_name, " + "employee.type AS employee_type, " + "manager.manager_data AS manager_manager_data " + "FROM employee JOIN manager ON employee.id = manager.id " + "JOIN (employee AS employee_1 JOIN engineer AS engineer_1 " + "ON employee_1.id = engineer_1.id) " + "ON engineer_1.manager_id = manager.id " + "WHERE employee.type IN (:type_1)", + ) + + def test_single_inh_subclass_join_wpoly_joined_inh_subclass(self): + Boss = self.classes.Boss + poly = with_polymorphic( + self.classes.Employee, + [self.classes.Boss, self.classes.Manager, self.classes.Engineer], + self._with_poly_fixture(), + ) + + s = Session() + + q = s.query(Boss).join( + poly.Engineer, poly.Engineer.manager_id == Boss.id + ) + + self.assert_compile( + q, + "SELECT manager.id AS manager_id, employee.id AS employee_id, " + "employee.name AS employee_name, employee.type AS employee_type, " + "manager.manager_data AS manager_manager_data " + "FROM employee JOIN manager ON employee.id = manager.id " + "JOIN (SELECT employee.id AS employee_id, " + "employee.type AS employee_type, employee.name AS employee_name, " + "manager.manager_data AS manager_manager_data, " + "NULL AS engineer_info, NULL AS manager_id " + "FROM employee JOIN manager ON employee.id = manager.id " + "UNION ALL " + "SELECT employee.id AS employee_id, " + "employee.type AS employee_type, employee.name AS employee_name, " + "NULL AS manager_data, " + "engineer.engineer_info AS engineer_engineer_info, " + "engineer.manager_id AS engineer_manager_id " + "FROM employee " + "JOIN engineer ON employee.id = engineer.id) AS anon_1 " + "ON anon_1.manager_id = manager.id " + "WHERE employee.type IN (:type_1)", + ) + + def test_joined_inh_subclass_join_single_inh_subclass(self): + Engineer = self.classes.Engineer + Boss = self.classes.Boss + s = Session() + + q = s.query(Engineer).join(Boss, Engineer.manager_id == Boss.id) + + self.assert_compile( + q, + "SELECT engineer.id AS engineer_id, employee.id AS employee_id, " + "employee.name AS employee_name, employee.type AS employee_type, " + "engineer.engineer_info AS engineer_engineer_info, " + "engineer.manager_id AS engineer_manager_id " + "FROM employee JOIN engineer ON employee.id = engineer.id " + "JOIN (employee AS employee_1 JOIN manager AS manager_1 " + "ON employee_1.id = manager_1.id) " + "ON engineer.manager_id = manager_1.id " + "AND employee_1.type IN (:type_1)", + ) + + class EagerDefaultEvalTest(fixtures.DeclarativeMappedTest): @classmethod def setup_classes(cls, with_polymorphic=None, include_sub_defaults=False): |