from sqlalchemy import util
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import clear_mappers
+from sqlalchemy.orm import column_property
+from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import create_session
from sqlalchemy.orm import join
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import Query
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
+from sqlalchemy.orm import with_polymorphic
from sqlalchemy.orm.interfaces import MANYTOONE
from sqlalchemy.testing import AssertsExecutionResults
from sqlalchemy.testing import eq_
for data in (True, False):
_fn = _generate_test(jointype, data)
setattr(RelationshipTest3, _fn.__name__, _fn)
-del func
+del _fn
class RelationshipTest4(fixtures.MappedTest):
"(a AS a_1 JOIN b AS b_1 ON a_1.id = b_1.id) "
"ON a.id = b_1.id WHERE b_1.x = :x_1",
)
+
+
+class CorrelateExceptWPolyAdaptTest(
+ fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
+):
+ # test [ticket:4537]'s test case.
+
+ run_create_tables = run_deletes = None
+ run_setup_classes = run_setup_mappers = run_define_tables = "each"
+ __dialect__ = "default"
+
+ def _fixture(self, use_correlate_except):
+
+ Base = self.DeclarativeBasic
+
+ class Superclass(Base):
+ __tablename__ = "s1"
+ id = Column(Integer, primary_key=True)
+ common_id = Column(ForeignKey("c.id"))
+ common_relationship = relationship(
+ "Common", uselist=False, innerjoin=True, lazy="noload"
+ )
+ discriminator_field = Column(String)
+ __mapper_args__ = {
+ "polymorphic_identity": "superclass",
+ "polymorphic_on": discriminator_field,
+ }
+
+ class Subclass(Superclass):
+ __tablename__ = "s2"
+ id = Column(ForeignKey("s1.id"), primary_key=True)
+ __mapper_args__ = {"polymorphic_identity": "subclass"}
+
+ class Common(Base):
+ __tablename__ = "c"
+ id = Column(Integer, primary_key=True)
+
+ if use_correlate_except:
+ num_superclass = column_property(
+ select([func.count(Superclass.id)])
+ .where(Superclass.common_id == id)
+ .correlate_except(Superclass)
+ )
+
+ if not use_correlate_except:
+ Common.num_superclass = column_property(
+ select([func.count(Superclass.id)])
+ .where(Superclass.common_id == Common.id)
+ .correlate(Common)
+ )
+
+ return Common, Superclass
+
+ def test_poly_query_on_correlate(self):
+ Common, Superclass = self._fixture(False)
+
+ poly = with_polymorphic(Superclass, "*")
+
+ s = Session()
+ q = (
+ s.query(poly)
+ .options(contains_eager(poly.common_relationship))
+ .join(poly.common_relationship)
+ .filter(Common.id == 1)
+ )
+
+ # note the order of c.id, subquery changes based on if we
+ # used correlate or correlate_except; this is only with the
+ # patch in place. Not sure why this happens.
+ self.assert_compile(
+ q,
+ "SELECT c.id AS c_id, (SELECT count(s1.id) AS count_1 "
+ "FROM s1 LEFT OUTER JOIN s2 ON s1.id = s2.id "
+ "WHERE s1.common_id = c.id) AS anon_1, "
+ "s1.id AS s1_id, "
+ "s1.common_id AS s1_common_id, "
+ "s1.discriminator_field AS s1_discriminator_field, "
+ "s2.id AS s2_id FROM s1 "
+ "LEFT OUTER JOIN s2 ON s1.id = s2.id "
+ "JOIN c ON c.id = s1.common_id WHERE c.id = :id_1",
+ )
+
+ def test_poly_query_on_correlate_except(self):
+ Common, Superclass = self._fixture(True)
+
+ poly = with_polymorphic(Superclass, "*")
+
+ s = Session()
+ q = (
+ s.query(poly)
+ .options(contains_eager(poly.common_relationship))
+ .join(poly.common_relationship)
+ .filter(Common.id == 1)
+ )
+
+ # c.id, subquery are reversed.
+ self.assert_compile(
+ q,
+ "SELECT (SELECT count(s1.id) AS count_1 "
+ "FROM s1 LEFT OUTER JOIN s2 ON s1.id = s2.id "
+ "WHERE s1.common_id = c.id) AS anon_1, "
+ "c.id AS c_id, s1.id AS s1_id, "
+ "s1.common_id AS s1_common_id, "
+ "s1.discriminator_field AS s1_discriminator_field, "
+ "s2.id AS s2_id FROM s1 "
+ "LEFT OUTER JOIN s2 ON s1.id = s2.id "
+ "JOIN c ON c.id = s1.common_id WHERE c.id = :id_1",
+ )
"AS anon_1 FROM table1 AS t1alias",
)
+ def test_correlate_except_on_clone(self):
+ # test [ticket:4537]'s issue
+
+ t1alias = t1.alias("t1alias")
+ j = t1.join(t1alias, t1.c.col1 == t1alias.c.col2)
+
+ vis = sql_util.ClauseAdapter(j)
+
+ # "control" subquery - uses correlate which has worked w/ adaption
+ # for a long time
+ control_s = (
+ select([t2.c.col1])
+ .where(t2.c.col1 == t1.c.col1)
+ .correlate(t2)
+ .as_scalar()
+ )
+
+ # test subquery - given only t1 and t2 in the enclosing selectable,
+ # will do the same thing as the "control" query since the correlation
+ # works out the same
+ s = (
+ select([t2.c.col1])
+ .where(t2.c.col1 == t1.c.col1)
+ .correlate_except(t1)
+ .as_scalar()
+ )
+
+ # use both subqueries in statements
+ control_stmt = select([control_s, t1.c.col1, t2.c.col1]).select_from(
+ t1.join(t2, t1.c.col1 == t2.c.col1)
+ )
+
+ stmt = select([s, t1.c.col1, t2.c.col1]).select_from(
+ t1.join(t2, t1.c.col1 == t2.c.col1)
+ )
+ # they are the same
+ self.assert_compile(
+ control_stmt,
+ "SELECT "
+ "(SELECT table2.col1 FROM table1 "
+ "WHERE table2.col1 = table1.col1) AS anon_1, "
+ "table1.col1, table2.col1 "
+ "FROM table1 "
+ "JOIN table2 ON table1.col1 = table2.col1",
+ )
+ self.assert_compile(
+ stmt,
+ "SELECT "
+ "(SELECT table2.col1 FROM table1 "
+ "WHERE table2.col1 = table1.col1) AS anon_1, "
+ "table1.col1, table2.col1 "
+ "FROM table1 "
+ "JOIN table2 ON table1.col1 = table2.col1",
+ )
+
+ # now test against the adaption of "t1" into "t1 JOIN t1alias".
+ # note in the control case, we aren't actually testing that
+ # Select is processing the "correlate" list during the adaption
+ # since we aren't adapting the "correlate"
+ self.assert_compile(
+ vis.traverse(control_stmt),
+ "SELECT "
+ "(SELECT table2.col1 FROM "
+ "table1 JOIN table1 AS t1alias ON table1.col1 = t1alias.col2 "
+ "WHERE table2.col1 = table1.col1) AS anon_1, "
+ "table1.col1, table2.col1 "
+ "FROM table1 JOIN table1 AS t1alias ON table1.col1 = t1alias.col2 "
+ "JOIN table2 ON table1.col1 = table2.col1",
+ )
+
+ # but here, correlate_except() does have the thing we're adapting
+ # so whatever is in there has to be expanded out to include
+ # the adaptation target, in this case "t1 JOIN t1alias".
+ self.assert_compile(
+ vis.traverse(stmt),
+ "SELECT "
+ "(SELECT table2.col1 FROM "
+ "table1 JOIN table1 AS t1alias ON table1.col1 = t1alias.col2 "
+ "WHERE table2.col1 = table1.col1) AS anon_1, "
+ "table1.col1, table2.col1 "
+ "FROM table1 JOIN table1 AS t1alias ON table1.col1 = t1alias.col2 "
+ "JOIN table2 ON table1.col1 = table2.col1",
+ )
+
@testing.fails_on_everything_except()
def test_joins_dont_adapt(self):
# adapting to a join, i.e. ClauseAdapter(t1.join(t2)), doesn't
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
