import datetime
+from typing import List
+from typing import Optional
import sqlalchemy as sa
from sqlalchemy import and_
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import foreign
from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import remote
from sqlalchemy.orm import selectinload
)
+class SecondaryCoversParentFlagTestM2M(fixtures.DeclarativeMappedTest):
+ def test_false_no_secondary(self):
+ Base = declarative_base()
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ bs: Mapped[list["B"]] = relationship("B", back_populates="a")
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ a_id: Mapped[int] = mapped_column(ForeignKey("a.id"))
+ a: Mapped["A"] = relationship("A", back_populates="bs")
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+ def test_joins_only_on_unique_non_pk(self):
+ Base = declarative_base()
+
+ atob = Table(
+ "atob",
+ Base.metadata,
+ Column("a_code", ForeignKey("a.code")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ code: Mapped[str] = mapped_column(unique=True)
+ bs: Mapped[list["B"]] = relationship("B", secondary=atob)
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+ def test_joins_only_on_unique_non_pk_to_subclass(self):
+ Base = declarative_base()
+
+ a_b = Table(
+ "a_b",
+ Base.metadata,
+ Column(
+ "a_code", String, ForeignKey("a_child.code"), primary_key=True
+ ),
+ Column("b_id", Integer, ForeignKey("b.id"), primary_key=True),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ type: Mapped[str]
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "a",
+ }
+
+ class AChild(A):
+ __tablename__ = "a_child"
+ id: Mapped[int] = mapped_column(
+ ForeignKey("a.id"), primary_key=True
+ )
+ code: Mapped[str] = mapped_column(unique=True)
+ bs: Mapped[list["B"]] = relationship(secondary=a_b)
+ __mapper_args__ = {"polymorphic_identity": "a_child"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = AChild.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+ def test_joins_from_inherited_subclass(self):
+ Base = self.DeclarativeBasic
+
+ a_b = Table(
+ "a_b",
+ Base.metadata,
+ Column(
+ "a_child_id",
+ String,
+ ForeignKey("a_child.id"),
+ primary_key=True,
+ ),
+ Column("b_id", Integer, ForeignKey("b.id"), primary_key=True),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ type: Mapped[str]
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "a",
+ }
+
+ class AChild(A):
+ __tablename__ = "a_child"
+ id: Mapped[int] = mapped_column(
+ ForeignKey("a.id"), primary_key=True
+ )
+ code: Mapped[str] = mapped_column(unique=True)
+ bs: Mapped[list["B"]] = relationship(secondary=a_b)
+ __mapper_args__ = {"polymorphic_identity": "a_child"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = AChild.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is True
+
+ def test_joins_from_parent_with_inheritance(self):
+ Base = declarative_base()
+
+ a_b = Table(
+ "a_b",
+ Base.metadata,
+ Column("a_id", Integer, ForeignKey("a.id"), primary_key=True),
+ Column("b_id", Integer, ForeignKey("b.id"), primary_key=True),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ type: Mapped[str] = mapped_column()
+ bs = relationship("B", secondary=a_b)
+ __mapper_args__ = {
+ "polymorphic_on": type,
+ "polymorphic_identity": "parent",
+ }
+
+ class AChild(A):
+ __tablename__ = "a_child"
+ id: Mapped[int] = mapped_column(
+ ForeignKey("a.id"), primary_key=True
+ )
+ code: Mapped[str] = mapped_column(unique=True)
+ __mapper_args__ = {"polymorphic_identity": "child"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is True
+
+ def test_joins_from_superclass_relationship_from_subclass(self):
+ """
+ This is a special case where relationship bs is for
+ AChild, and AChild's pk is a fk from parent A. But
+ for table a_b the keys come from parent A and B.
+
+ We are expecting the results to be False since
+ it should fail the `issubset` evaluation between AChild
+ and B.
+
+ When forcing True to allow omit_join=True for this example,
+ the IN statement still works and gets optimized since
+ AChild.id = A.id
+ """
+ Base = declarative_base()
+
+ a_b = Table(
+ "a_b",
+ Base.metadata,
+ Column("a_id", Integer, ForeignKey("a.id"), primary_key=True),
+ Column("b_id", Integer, ForeignKey("b.id"), primary_key=True),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ type: Mapped[str] = mapped_column()
+ __mapper_args__ = {
+ "polymorphic_on": type,
+ "polymorphic_identity": "parent",
+ }
+
+ class AChild(A):
+ __tablename__ = "a_child"
+ id: Mapped[int] = mapped_column(
+ ForeignKey("a.id"), primary_key=True
+ )
+ code: Mapped[str] = mapped_column(unique=True)
+ bs = relationship("B", secondary=a_b)
+ __mapper_args__ = {"polymorphic_identity": "child"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = AChild.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+ def test_joins_from_composite_pk(self):
+ Base = declarative_base()
+
+ association_table = Table(
+ "a_b",
+ Base.metadata,
+ Column("a_id1", Integer, ForeignKey("a.id1")),
+ Column("a_id2", Integer, ForeignKey("a.id2")),
+ Column("b_id", Integer, ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id1: Mapped[int] = mapped_column(primary_key=True)
+ id2: Mapped[int] = mapped_column(primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=lambda: and_(
+ A.id1 == association_table.c.a_id1,
+ A.id2 == association_table.c.a_id2,
+ ),
+ secondaryjoin=lambda: B.id == association_table.c.b_id,
+ )
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is True
+
+ def test_only_partial_of_composite_pk(self):
+ Base = declarative_base()
+
+ association_table = Table(
+ "a_b",
+ Base.metadata,
+ Column("a_id1", Integer, ForeignKey("a.id1")),
+ Column("b_id", Integer, ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id1: Mapped[int] = mapped_column(primary_key=True)
+ id2: Mapped[int] = mapped_column(primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=lambda: A.id1 == association_table.c.a_id1,
+ secondaryjoin=lambda: B.id == association_table.c.b_id,
+ )
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+ def test_simple(self):
+ Base = declarative_base()
+
+ atob = Table(
+ "atob",
+ Base.metadata,
+ Column("a_id", ForeignKey("a.id")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ bs: Mapped[list["B"]] = relationship("B", secondary=atob)
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is True
+
+ def test_bidirectional(self):
+ Base = declarative_base()
+
+ atob = Table(
+ "atob",
+ Base.metadata,
+ Column("a_id", ForeignKey("a.id")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ bs: Mapped[list["B"]] = relationship(
+ "B", secondary=atob, back_populates="as_"
+ )
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ as_: Mapped[list["A"]] = relationship(
+ "A", secondary=atob, back_populates="bs"
+ )
+
+ join_cond = A.bs.property._join_condition
+ a_flag = join_cond.secondary_covers_parent_primary_key
+
+ join_cond = B.as_.property._join_condition
+ b_flag = join_cond.secondary_covers_parent_primary_key
+
+ assert a_flag is True
+ assert b_flag is True
+
+ def test_viewonly_association_table(self):
+ """
+ Example pulled from Sqlalchemy documentation, with the only
+ changes was to include "viewonly=True" to the secondary
+ relationship.
+ """
+ Base = declarative_base()
+
+ class Association(Base):
+ __tablename__ = "association_table"
+
+ left_id: Mapped[int] = mapped_column(
+ ForeignKey("left_table.id"), primary_key=True
+ )
+ right_id: Mapped[int] = mapped_column(
+ ForeignKey("right_table.id"), primary_key=True
+ )
+ extra_data: Mapped[Optional[str]]
+
+ # association between Association -> Child
+ child: Mapped["Child"] = relationship(
+ back_populates="parent_associations"
+ )
+
+ # association between Association -> Parent
+ parent: Mapped["Parent"] = relationship(
+ back_populates="child_associations"
+ )
+
+ class Parent(Base):
+ __tablename__ = "left_table"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ # many-to-many relationship to Child,
+ # bypassing the `Association` class
+ children: Mapped[List["Child"]] = relationship(
+ secondary="association_table",
+ back_populates="parents",
+ viewonly=True,
+ )
+
+ # association between Parent -> Association -> Child
+ child_associations: Mapped[List["Association"]] = relationship(
+ back_populates="parent"
+ )
+
+ class Child(Base):
+ __tablename__ = "right_table"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ # many-to-many relationship to Parent,
+ # bypassing the `Association` class
+ parents: Mapped[List["Parent"]] = relationship(
+ secondary="association_table",
+ back_populates="children",
+ viewonly=True,
+ )
+
+ # association between Child -> Association -> Parent
+ parent_associations: Mapped[List["Association"]] = relationship(
+ back_populates="child"
+ )
+
+ join_cond = Parent.children.property._join_condition
+ parent_child_flag = join_cond.secondary_covers_parent_primary_key
+
+ join_cond = Child.parents.property._join_condition
+ child_parent_flag = join_cond.secondary_covers_parent_primary_key
+
+ assert parent_child_flag is True
+ assert child_parent_flag is True
+
+ def test_mapper_only_pk_true(self):
+ """secondary joins on the mapper-specified primary_key column,
+ not the table's actual primary key."""
+ Base = declarative_base()
+
+ atob = Table(
+ "atob",
+ Base.metadata,
+ Column("a_code", ForeignKey("a.code")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ code: Mapped[str] = mapped_column(unique=True)
+ bs: Mapped[list["B"]] = relationship("B", secondary=atob)
+ __mapper_args__ = {"primary_key": "code"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is True
+
+ def test_mapper_only_pk_false(self):
+ """secondary does not join on the mapper-specified primary_key
+ column, so the flag should be False."""
+ Base = declarative_base()
+
+ atob = Table(
+ "atob",
+ Base.metadata,
+ Column("a_id", ForeignKey("a.id")),
+ Column("b_id", ForeignKey("b.id")),
+ )
+
+ class A(Base):
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ code: Mapped[str] = mapped_column(unique=True)
+ bs: Mapped[list["B"]] = relationship("B", secondary=atob)
+ __mapper_args__ = {"primary_key": "code"}
+
+ class B(Base):
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ join_cond = A.bs.property._join_condition
+ flag = join_cond.secondary_covers_parent_primary_key
+
+ assert flag is False
+
+
class ActiveHistoryFlagTest(_fixtures.FixtureTest):
run_inserts = None
run_deletes = None
params=[{"id_1": 2}],
),
CompiledSQL(
- "SELECT a_1.id, b.id FROM a AS a_1 JOIN "
+ "SELECT anon_1.aid, b.id FROM "
"(SELECT a.id AS aid, b.id AS id FROM a JOIN b ON a.b_ids "
"LIKE (:id_1 || b.id || :param_1)) AS anon_1 "
- "ON a_1.id = anon_1.aid JOIN b ON b.id = anon_1.id "
- "WHERE a_1.id IN (__[POSTCOMPILE_primary_keys])",
+ "JOIN b ON b.id = anon_1.id "
+ "WHERE anon_1.aid IN (__[POSTCOMPILE_primary_keys])",
params=[{"id_1": "%", "param_1": "%", "primary_keys": [2]}],
),
)
import sqlalchemy as sa
+from sqlalchemy import and_
from sqlalchemy import bindparam
+from sqlalchemy import Boolean
from sqlalchemy import ForeignKey
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Integer
from sqlalchemy.orm import with_polymorphic
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assert_warns
+from sqlalchemy.testing import AssertsExecutionResults
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
)
+class M2MOmitJoinTest(
+ fixtures.TestBase, AssertsExecutionResults, testing.AssertsCompiledSQL
+):
+ __dialect__ = "default"
+
+ @testing.fixture
+ def simple_m2m(self, decl_base, connection):
+ association_table = Table(
+ "a_b",
+ decl_base.metadata,
+ Column("a_id", Integer, ForeignKey("a.id")),
+ Column("b_id", Integer, ForeignKey("b.id")),
+ )
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ bs = relationship("B", secondary=association_table)
+ bs_no_omit_join = relationship(
+ "B",
+ secondary=association_table,
+ omit_join=False,
+ overlaps="bs",
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+
+ decl_base.metadata.create_all(connection)
+
+ with Session(connection) as session:
+ a1 = A(id=1)
+ a2 = A(id=2)
+ b1 = B(id=1)
+ b2 = B(id=2)
+ a1.bs = [b1, b2]
+ a2.bs = [b1]
+ session.add_all([a1, a2, b1, b2])
+ session.commit()
+
+ return A
+
+ @testing.fixture
+ def symmetric_composite_m2m(self, decl_base, connection):
+ association_table = Table(
+ "a_b",
+ decl_base.metadata,
+ Column("a_id1", Integer, ForeignKey("a.id1")),
+ Column("a_id2", Integer, ForeignKey("a.id2")),
+ Column("b_id1", Integer, ForeignKey("b.id1")),
+ Column("b_id2", Integer, ForeignKey("b.id2")),
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id1 = Column(Integer, primary_key=True)
+ id2 = Column(Integer, primary_key=True)
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id1 = Column(Integer, primary_key=True)
+ id2 = Column(Integer, primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=lambda: and_(
+ A.id1 == association_table.c.a_id1,
+ A.id2 == association_table.c.a_id2,
+ ),
+ secondaryjoin=lambda: and_(
+ B.id1 == association_table.c.b_id1,
+ B.id2 == association_table.c.b_id2,
+ ),
+ )
+ bs_no_omit_join = relationship(
+ "B",
+ secondary=association_table,
+ omit_join=False,
+ overlaps="bs",
+ primaryjoin=lambda: and_(
+ A.id1 == association_table.c.a_id1,
+ A.id2 == association_table.c.a_id2,
+ ),
+ secondaryjoin=lambda: and_(
+ B.id1 == association_table.c.b_id1,
+ B.id2 == association_table.c.b_id2,
+ ),
+ )
+
+ decl_base.metadata.create_all(connection)
+
+ with Session(connection) as session:
+ a1 = A(id1=1, id2=1)
+ a2 = A(id1=1, id2=2)
+ b1 = B(id1=1, id2=1)
+ b2 = B(id1=1, id2=2)
+ a1.bs = [b1, b2]
+ a2.bs = [b1]
+ session.add_all([a1, a2, b1, b2])
+ session.commit()
+
+ return A
+
+ @testing.fixture
+ def asymmetric_composite_m2m(self, decl_base, connection):
+ association_table = Table(
+ "a_b",
+ decl_base.metadata,
+ Column("a_id1", Integer, ForeignKey("a.id1")),
+ Column("a_id2", Integer, ForeignKey("a.id2")),
+ Column("b_id", Integer, ForeignKey("b.id")),
+ )
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id1 = Column(Integer, primary_key=True)
+ id2 = Column(Integer, primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=lambda: and_(
+ A.id1 == association_table.c.a_id1,
+ A.id2 == association_table.c.a_id2,
+ ),
+ secondaryjoin=(lambda: B.id == association_table.c.b_id),
+ )
+ bs_no_omit_join = relationship(
+ "B",
+ secondary=association_table,
+ omit_join=False,
+ overlaps="bs",
+ primaryjoin=lambda: and_(
+ A.id1 == association_table.c.a_id1,
+ A.id2 == association_table.c.a_id2,
+ ),
+ secondaryjoin=(lambda: B.id == association_table.c.b_id),
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+
+ decl_base.metadata.create_all(connection)
+
+ with Session(connection) as session:
+ a1 = A(id1=1, id2=1)
+ a2 = A(id1=1, id2=2)
+ b1 = B(id=1)
+ b2 = B(id=2)
+ b3 = B(id=3)
+ a1.bs = [b1, b2, b3]
+ a2.bs = [b2, b3]
+ session.add_all([a1, a2, b1, b2, b3])
+ session.commit()
+
+ return A
+
+ @testing.fixture
+ def reverse_asymmetric_composite_m2m(self, decl_base, connection):
+ association_table = Table(
+ "a_b",
+ decl_base.metadata,
+ Column("a_id", Integer, ForeignKey("a.id")),
+ Column("b_id1", Integer, ForeignKey("b.id1")),
+ Column("b_id2", Integer, ForeignKey("b.id2")),
+ )
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=(lambda: A.id == association_table.c.a_id),
+ secondaryjoin=lambda: and_(
+ B.id1 == association_table.c.b_id1,
+ B.id2 == association_table.c.b_id2,
+ ),
+ )
+ bs_no_omit_join = relationship(
+ "B",
+ secondary=association_table,
+ omit_join=False,
+ overlaps="bs",
+ primaryjoin=(lambda: A.id == association_table.c.a_id),
+ secondaryjoin=lambda: and_(
+ B.id1 == association_table.c.b_id1,
+ B.id2 == association_table.c.b_id2,
+ ),
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id1 = Column(Integer, primary_key=True)
+ id2 = Column(Integer, primary_key=True)
+
+ decl_base.metadata.create_all(connection)
+
+ with Session(connection) as session:
+ a1 = A(id=1)
+ a2 = A(id=2)
+ b1 = B(id1=1, id2=1)
+ b2 = B(id1=1, id2=2)
+ b3 = B(id1=2, id2=1)
+ a1.bs = [b1, b2, b3]
+ a2.bs = [b2, b3]
+ session.add_all([a1, a2, b1, b2, b3])
+ session.commit()
+
+ return A
+
+ @testing.fixture
+ def filtered_secondaryjoin_m2m(self, decl_base, connection):
+ association_table = Table(
+ "a_b",
+ decl_base.metadata,
+ Column("a_id", Integer, ForeignKey("a.id")),
+ Column("b_id", Integer, ForeignKey("b.id")),
+ )
+
+ class A(decl_base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ bs = relationship(
+ "B",
+ secondary=association_table,
+ primaryjoin=(lambda: A.id == association_table.c.a_id),
+ secondaryjoin=lambda: and_(
+ B.id == association_table.c.b_id,
+ B.active == True, # noqa: E712
+ ),
+ )
+ bs_no_omit_join = relationship(
+ "B",
+ secondary=association_table,
+ omit_join=False,
+ overlaps="bs",
+ primaryjoin=(lambda: A.id == association_table.c.a_id),
+ secondaryjoin=lambda: and_(
+ B.id == association_table.c.b_id,
+ B.active == True, # noqa: E712
+ ),
+ )
+
+ class B(decl_base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ active = Column(Boolean, default=True)
+
+ decl_base.metadata.create_all(connection)
+
+ with Session(connection) as session:
+ a1 = A(id=1)
+ a2 = A(id=2)
+ b1 = B(id=1, active=True)
+ b2 = B(id=2, active=False)
+ b3 = B(id=3, active=True)
+ a1.bs = [b1, b2, b3]
+ a2.bs = [b2, b3]
+ session.add_all([a1, a2, b1, b2, b3])
+ session.commit()
+
+ return A
+
+ def test_simple_optimized(self, simple_m2m, connection):
+ A = simple_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A).options(selectinload(A.bs)).order_by(A.id)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL("SELECT a.id FROM a ORDER BY a.id", {}),
+ CompiledSQL(
+ "SELECT a_b.a_id, b.id "
+ "FROM a_b JOIN b ON b.id = a_b.b_id "
+ "WHERE a_b.a_id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ def test_simple_unoptimized(self, simple_m2m, connection):
+ A = simple_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs_no_omit_join))
+ .order_by(A.id)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL("SELECT a.id FROM a ORDER BY a.id", {}),
+ CompiledSQL(
+ "SELECT a_1.id, b.id "
+ "FROM a AS a_1 "
+ "JOIN a_b AS a_b_1 ON a_1.id = a_b_1.a_id "
+ "JOIN b ON b.id = a_b_1.b_id "
+ "WHERE a_1.id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ def test_symmetric_composite(self, symmetric_composite_m2m, connection):
+ A = symmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs))
+ .order_by(A.id1, A.id2)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id1, a.id2 " "FROM a ORDER BY a.id1, a.id2",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_b.a_id1, a_b.a_id2, b.id1, b.id2 "
+ "FROM a_b "
+ "JOIN b ON b.id1 = a_b.b_id1 "
+ "AND b.id2 = a_b.b_id2 "
+ "WHERE (a_b.a_id1, a_b.a_id2) IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [(1, 1), (1, 2)]},
+ ),
+ )
+
+ def test_symmetric_composite_unoptimized(
+ self, symmetric_composite_m2m, connection
+ ):
+ A = symmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs_no_omit_join))
+ .order_by(A.id1, A.id2)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id1, a.id2 " "FROM a ORDER BY a.id1, a.id2",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_1.id1, a_1.id2, b.id1, b.id2 "
+ "FROM a AS a_1 JOIN a_b AS a_b_1 ON "
+ "a_1.id1 = a_b_1.a_id1 "
+ "AND a_1.id2 = a_b_1.a_id2 "
+ "JOIN b ON b.id1 = a_b_1.b_id1 "
+ "AND b.id2 = a_b_1.b_id2 "
+ "WHERE (a_1.id1, a_1.id2) IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [(1, 1), (1, 2)]},
+ ),
+ )
+
+ def test_asymmetric_composite(self, asymmetric_composite_m2m, connection):
+ A = asymmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs))
+ .order_by(A.id1, A.id2)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id1, a.id2 FROM a ORDER BY a.id1, a.id2",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_b.a_id1, a_b.a_id2, b.id "
+ "FROM a_b JOIN b ON b.id = a_b.b_id "
+ "WHERE (a_b.a_id1, a_b.a_id2) IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [(1, 1), (1, 2)]},
+ ),
+ )
+
+ def test_asymmetric_composite_unoptimized(
+ self, asymmetric_composite_m2m, connection
+ ):
+ A = asymmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs_no_omit_join))
+ .order_by(A.id1, A.id2)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id1, a.id2 FROM a ORDER BY a.id1, a.id2",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_1.id1, a_1.id2, b.id "
+ "FROM a AS a_1 JOIN a_b AS a_b_1 "
+ "ON a_1.id1 = a_b_1.a_id1 "
+ "AND a_1.id2 = a_b_1.a_id2 "
+ "JOIN b ON b.id = a_b_1.b_id "
+ "WHERE (a_1.id1, a_1.id2) IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [(1, 1), (1, 2)]},
+ ),
+ )
+
+ def test_reverse_asymmetric_composite(
+ self, reverse_asymmetric_composite_m2m, connection
+ ):
+ A = reverse_asymmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A).options(selectinload(A.bs)).order_by(A.id)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id FROM a ORDER BY a.id",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_b.a_id, b.id1, b.id2 "
+ "FROM a_b "
+ "JOIN b ON b.id1 = a_b.b_id1 "
+ "AND b.id2 = a_b.b_id2 "
+ "WHERE a_b.a_id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ def test_reverse_asymmetric_composite_unoptimized(
+ self, reverse_asymmetric_composite_m2m, connection
+ ):
+ A = reverse_asymmetric_composite_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs_no_omit_join))
+ .order_by(A.id)
+ )
+ session.execute(statement).scalars().all()
+
+ self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id FROM a ORDER BY a.id",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_1.id, b.id1, b.id2 "
+ "FROM a AS a_1 "
+ "JOIN a_b AS a_b_1 ON a_1.id = a_b_1.a_id "
+ "JOIN b ON b.id1 = a_b_1.b_id1 "
+ "AND b.id2 = a_b_1.b_id2 "
+ "WHERE a_1.id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ def test_filtered_secondaryjoin(
+ self, filtered_secondaryjoin_m2m, connection
+ ):
+ A = filtered_secondaryjoin_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A).options(selectinload(A.bs)).order_by(A.id)
+ )
+ return session.execute(statement).scalars().all()
+
+ results = self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id FROM a ORDER BY a.id",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_b.a_id, b.id, b.active "
+ "FROM a_b "
+ "JOIN b ON b.id = a_b.b_id AND b.active = 1 "
+ "WHERE a_b.a_id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ eq_(sorted(b.id for b in results[0].bs), [1, 3])
+ eq_(sorted(b.id for b in results[1].bs), [3])
+
+ def test_filtered_secondaryjoin_unoptimized(
+ self, filtered_secondaryjoin_m2m, connection
+ ):
+ A = filtered_secondaryjoin_m2m
+
+ with Session(connection) as session:
+
+ def go():
+ statement = (
+ select(A)
+ .options(selectinload(A.bs_no_omit_join))
+ .order_by(A.id)
+ )
+ return session.execute(statement).scalars().all()
+
+ results = self.assert_sql_execution(
+ connection,
+ go,
+ CompiledSQL(
+ "SELECT a.id FROM a ORDER BY a.id",
+ {},
+ ),
+ CompiledSQL(
+ "SELECT a_1.id, b.id, b.active "
+ "FROM a AS a_1 "
+ "JOIN a_b AS a_b_1 ON a_1.id = a_b_1.a_id "
+ "JOIN b ON b.id = a_b_1.b_id AND b.active = 1 "
+ "WHERE a_1.id IN "
+ "(__[POSTCOMPILE_primary_keys])",
+ {"primary_keys": [1, 2]},
+ ),
+ )
+
+ eq_(
+ sorted(b.id for b in results[0].bs_no_omit_join),
+ [1, 3],
+ )
+ eq_(
+ sorted(b.id for b in results[1].bs_no_omit_join),
+ [3],
+ )
+
+
class SameNamePolymorphicTest(fixtures.DeclarativeMappedTest):
@classmethod
def setup_classes(cls):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
