From: Mike Bayer Date: Sat, 15 Jun 2013 19:20:48 +0000 (-0400) Subject: Fixed bug in polymorphic SQL generation where multiple joined-inheritance X-Git-Tag: rel_0_8_2~46 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=aee826890c7570fcf397c53794ff2b7418e24a97;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git Fixed bug in polymorphic SQL generation where multiple joined-inheritance entities against the same base class joined to each other as well would not track columns on the base table independently of each other if the string of joins were more than two entities long. Also in 0.8.2. [ticket:2759] Conflicts: doc/build/changelog/changelog_09.rst --- diff --git a/doc/build/changelog/changelog_08.rst b/doc/build/changelog/changelog_08.rst index 8e81ff1b94..ffb2a24d42 100644 --- a/doc/build/changelog/changelog_08.rst +++ b/doc/build/changelog/changelog_08.rst @@ -6,6 +6,15 @@ .. changelog:: :version: 0.8.2 + .. change:: + :tags: bug, orm + :tickets: 2759 + + Fixed bug in polymorphic SQL generation where multiple joined-inheritance + entities against the same base class joined to each other as well + would not track columns on the base table independently of each other if + the string of joins were more than two entities long. + .. change:: :tags: bug, engine :pullreq: 6 diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py index c2ec72c510..13699316fb 100644 --- a/lib/sqlalchemy/orm/query.py +++ b/lib/sqlalchemy/orm/query.py @@ -203,17 +203,22 @@ class Query(object): self._polymorphic_adapters.pop(m.local_table, None) def _adapt_polymorphic_element(self, element): + if "parententity" in element._annotations: + search = element._annotations['parententity'] + alias = self._polymorphic_adapters.get(search, None) + if alias: + return alias.adapt_clause(element) + if isinstance(element, expression.FromClause): search = element elif hasattr(element, 'table'): search = element.table else: - search = None + return None - if search is not None: - alias = self._polymorphic_adapters.get(search, None) - if alias: - return alias.adapt_clause(element) + alias = self._polymorphic_adapters.get(search, None) + if alias: + return alias.adapt_clause(element) def _adapt_col_list(self, cols): return [ diff --git a/test/orm/inheritance/test_relationship.py b/test/orm/inheritance/test_relationship.py index 36dbb7d27e..c566c5c494 100644 --- a/test/orm/inheritance/test_relationship.py +++ b/test/orm/inheritance/test_relationship.py @@ -8,7 +8,7 @@ from sqlalchemy.engine import default from sqlalchemy.testing import AssertsCompiledSQL, fixtures from sqlalchemy import testing from sqlalchemy.testing.schema import Table, Column -from sqlalchemy.testing import assert_raises, eq_ +from sqlalchemy.testing import assert_raises, eq_, is_ class Company(fixtures.ComparableEntity): pass @@ -1377,3 +1377,93 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest): "JOIN ep2 ON anon_1.anon_2_base2_id = ep2.base2_id" ) +class MultipleAdaptUsesEntityOverTableTest(AssertsCompiledSQL, fixtures.MappedTest): + __dialect__ = 'default' + run_create_tables = None + + @classmethod + def define_tables(cls, metadata): + Table('a', metadata, + Column('id', Integer, primary_key=True), + Column('name', String) + ) + Table('b', metadata, + Column('id', Integer, ForeignKey('a.id'), primary_key=True) + ) + Table('c', metadata, + Column('id', Integer, ForeignKey('a.id'), primary_key=True), + Column('bid', Integer, ForeignKey('b.id')) + ) + Table('d', metadata, + Column('id', Integer, ForeignKey('a.id'), primary_key=True), + Column('cid', Integer, ForeignKey('c.id')) + ) + + @classmethod + def setup_classes(cls): + class A(cls.Comparable): + pass + class B(A): + pass + class C(A): + pass + class D(A): + pass + + @classmethod + def setup_mappers(cls): + A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D + a, b, c, d = cls.tables.a, cls.tables.b, cls.tables.c, cls.tables.d + mapper(A, a) + mapper(B, b, inherits=A) + mapper(C, c, inherits=A) + mapper(D, d, inherits=A) + + def _two_join_fixture(self): + A, B, C, D = self.classes.A, self.classes.B, self.classes.C, self.classes.D + s = Session() + return s.query(B.name, C.name, D.name).select_from(B).\ + join(C, C.bid == B.id).\ + join(D, D.cid == C.id) + + def test_two_joins_adaption(self): + a, b, c, d = self.tables.a, self.tables.b, self.tables.c, self.tables.d + q = self._two_join_fixture() + + btoc = q._from_obj[0].left + + ac_adapted = btoc.right.element.left + c_adapted = btoc.right.element.right + + is_(ac_adapted.element, a) + is_(c_adapted.element, c) + + ctod = q._from_obj[0].right + ad_adapted = ctod.left + d_adapted = ctod.right + is_(ad_adapted.element, a) + is_(d_adapted.element, d) + + bname, cname, dname = q._entities + + b_name_adapted = bname._resolve_expr_against_query_aliases( + q, bname.column, None) + c_name_adapted = cname._resolve_expr_against_query_aliases( + q, cname.column, None) + d_name_adapted = dname._resolve_expr_against_query_aliases( + q, dname.column, None) + + assert bool(b_name_adapted == a.c.name) + assert bool(c_name_adapted == ac_adapted.c.name) + assert bool(d_name_adapted == ad_adapted.c.name) + + def test_two_joins_sql(self): + q = self._two_join_fixture() + self.assert_compile(q, + "SELECT a.name AS a_name, a_1.name AS a_1_name, " + "a_2.name AS a_2_name " + "FROM a JOIN b ON a.id = b.id JOIN " + "(a AS a_1 JOIN c AS c_1 ON a_1.id = c_1.id) ON c_1.bid = b.id " + "JOIN (a AS a_2 JOIN d AS d_1 ON a_2.id = d_1.id) " + "ON d_1.cid = c_1.id" + )