self.support_sync = support_sync
self.can_be_synced_fn = can_be_synced_fn
self._determine_joins()
+ self._sanitize_joins()
self._annotate_fks()
self._annotate_remote()
self._annotate_local()
log.info('%s relationship direction %s', self.prop,
self.direction)
+ def _sanitize_joins(self):
+ """remove the parententity annotation from our join conditions which
+ can leak in here based on some declarative patterns and maybe others.
+
+ We'd want to remove "parentmapper" also, but apparently there's
+ an exotic use case in _join_fixture_inh_selfref_w_entity
+ that relies upon it being present, see :ticket:`3364`.
+
+ """
+
+ self.primaryjoin = _deep_deannotate(
+ self.primaryjoin, values=("parententity",))
+ if self.secondaryjoin is not None:
+ self.secondaryjoin = _deep_deannotate(
+ self.secondaryjoin, values=("parententity",))
+
def _determine_joins(self):
"""Determine the 'primaryjoin' and 'secondaryjoin' attributes,
if not passed to the constructor already.
from sqlalchemy import (
testing, null, exists, text, union, literal, literal_column, func, between,
Unicode, desc, and_, bindparam, select, distinct, or_, collate, insert,
- Integer, String, Boolean, exc as sa_exc, util, cast, MetaData)
+ Integer, String, Boolean, exc as sa_exc, util, cast, MetaData, ForeignKey)
from sqlalchemy.sql import operators, expression
from sqlalchemy import column, table
from sqlalchemy.engine import default
)
+class HasAnyTest(
+ fixtures.DeclarativeMappedTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class D(Base):
+ __tablename__ = 'd'
+ id = Column(Integer, primary_key=True)
+
+ class C(Base):
+ __tablename__ = 'c'
+ id = Column(Integer, primary_key=True)
+ d_id = Column(ForeignKey(D.id))
+
+ bs = relationship("B", back_populates="c")
+
+ b_d = Table(
+ 'b_d', Base.metadata,
+ Column('bid', ForeignKey('b.id')),
+ Column('did', ForeignKey('d.id'))
+ )
+
+ # note we are using the ForeignKey pattern identified as a bug
+ # in [ticket:4367]
+ class B(Base):
+ __tablename__ = 'b'
+ id = Column(Integer, primary_key=True)
+ c_id = Column(ForeignKey(C.id))
+
+ c = relationship("C", back_populates="bs")
+
+ d = relationship("D", secondary=b_d)
+
+ class A(Base):
+ __tablename__ = 'a'
+ id = Column(Integer, primary_key=True)
+ b_id = Column(ForeignKey(B.id))
+
+ def test_has_many_to_one(self):
+ B, C = self.classes("B", "C")
+ s = Session()
+ self.assert_compile(
+ s.query(B).filter(B.c.has(C.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id FROM b WHERE "
+ "EXISTS (SELECT 1 FROM c WHERE c.id = b.c_id AND c.id = :id_1)"
+ )
+
+ def test_any_many_to_many(self):
+ B, D = self.classes("B", "D")
+ s = Session()
+ self.assert_compile(
+ s.query(B).filter(B.d.any(D.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id FROM b WHERE "
+ "EXISTS (SELECT 1 FROM b_d, d WHERE b.id = b_d.bid "
+ "AND d.id = b_d.did AND d.id = :id_1)"
+ )
+
+ def test_any_one_to_many(self):
+ B, C = self.classes("B", "C")
+ s = Session()
+ self.assert_compile(
+ s.query(C).filter(C.bs.any(B.id == 1)),
+ "SELECT c.id AS c_id, c.d_id AS c_d_id FROM c WHERE "
+ "EXISTS (SELECT 1 FROM b WHERE c.id = b.c_id AND b.id = :id_1)"
+ )
+
+ def test_any_many_to_many_doesnt_overcorrelate(self):
+ B, D = self.classes("B", "D")
+ s = Session()
+
+ self.assert_compile(
+ s.query(B).join(B.d).filter(B.d.any(D.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id FROM "
+ "b JOIN b_d AS b_d_1 ON b.id = b_d_1.bid "
+ "JOIN d ON d.id = b_d_1.did WHERE "
+ "EXISTS (SELECT 1 FROM b_d, d WHERE b.id = b_d.bid "
+ "AND d.id = b_d.did AND d.id = :id_1)"
+ )
+
+ def test_has_doesnt_overcorrelate(self):
+ B, C = self.classes("B", "C")
+ s = Session()
+
+ self.assert_compile(
+ s.query(B).join(B.c).filter(B.c.has(C.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id "
+ "FROM b JOIN c ON c.id = b.c_id "
+ "WHERE EXISTS "
+ "(SELECT 1 FROM c WHERE c.id = b.c_id AND c.id = :id_1)"
+ )
+
+ def test_has_doesnt_get_aliased_join_subq(self):
+ B, C = self.classes("B", "C")
+ s = Session()
+
+ self.assert_compile(
+ s.query(B).join(B.c, aliased=True).filter(B.c.has(C.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id "
+ "FROM b JOIN c AS c_1 ON c_1.id = b.c_id "
+ "WHERE EXISTS "
+ "(SELECT 1 FROM c WHERE c.id = b.c_id AND c.id = :id_1)"
+ )
+
+ def test_any_many_to_many_doesnt_get_aliased_join_subq(self):
+ B, D = self.classes("B", "D")
+ s = Session()
+
+ self.assert_compile(
+ s.query(B).join(B.d, aliased=True).filter(B.d.any(D.id == 1)),
+ "SELECT b.id AS b_id, b.c_id AS b_c_id "
+ "FROM b JOIN b_d AS b_d_1 ON b.id = b_d_1.bid "
+ "JOIN d AS d_1 ON d_1.id = b_d_1.did "
+ "WHERE EXISTS "
+ "(SELECT 1 FROM b_d, d WHERE b.id = b_d.bid "
+ "AND d.id = b_d.did AND d.id = :id_1)"
+ )
+
+
class HasMapperEntitiesTest(QueryTest):
def test_entity(self):
User = self.classes.User
from sqlalchemy.testing import assert_raises_message, eq_, \
AssertsCompiledSQL, is_
from sqlalchemy.testing import fixtures
-from sqlalchemy.orm import relationships, foreign, remote
+from sqlalchemy.orm import relationships, foreign, remote, relationship
from sqlalchemy import MetaData, Table, Column, ForeignKey, Integer, \
select, ForeignKeyConstraint, exc, func, and_, String, Boolean
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
local_selectable = self.base.join(self.sub)
remote_selectable = self.base.join(self.sub_w_sub_rel)
+ # note this test requires that "parentmapper" annotation is
+ # present in the columns ahead of time
+
sub_w_sub_rel__sub_id = self.sub_w_sub_rel.c.sub_id._annotate(
{'parentmapper': prop.mapper})
sub__id = self.sub.c.id._annotate({'parentmapper': prop.parent})
":param_1 = selfref.sid OR selfref.sid = :param_1",
checkparams={'param_1': None}
)
+
+
+class DeannotateCorrectlyTest(fixtures.TestBase):
+ def test_pj_deannotates(self):
+ from sqlalchemy.ext.declarative import declarative_base
+ Base = declarative_base()
+
+ class A(Base):
+ __tablename__ = 'a'
+ id = Column(Integer, primary_key=True)
+
+ class B(Base):
+ __tablename__ = 'b'
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey(A.id))
+ a = relationship(A)
+
+ eq_(
+ B.a.property.primaryjoin.left._annotations,
+ {"parentmapper": A.__mapper__, "remote": True}
+ )
+ eq_(
+ B.a.property.primaryjoin.right._annotations,
+ {'foreign': True, 'local': True, 'parentmapper': B.__mapper__}
+ )
+
+
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
