Sqlalchemy: Resolve cartesian product warning for many-to-many relationship with custom primaryjoin

Let's say I have the following database schema:

class A(Base):
    __tablename__ = "a_table"

    id = Column(Integer, primary_key=True, autoincrement=True)
    version = Column(Integer, primary_key=True, default=1)

    # More columns...

    bs = relationship(
        "B", secondary="a2b_table", back_populates="as"
    )

class B(Base):
    __tablename__ = "b_table"

    id = Column(Integer, primary_key=True)

    as = relationship(
        A, secondary="a2b_table", back_populates="bs"
    )

class A2B(Base):
    __tablename__ = "a2b_table"

    a_id = Column(
        Integer(),
        primary_key=True,
    )
    a_version = Column(
        Integer,
        primary_key=True,
    )
    b_id = sa.Column(
        Integer,
        ForeignKey("b.id", name="b_fk"),
        primary_key=True,
    )

    __table_args__ = (
        ForeignKeyConstraint(
            [a_id, a_version],
            [A.id, A.version],
            name="a_fk",
        ),
        {},
    )

Each A is identified by an id and can have multiple versions. If something changes in the columns of A (the ones not shown), I produce a new A with the same id and version+1. The relationship bs gives me all instances of B that are associated with a specific version of an A.

The problem is, that the relationship as gives me all versions of each A that is associated with a specific B. Instead, I want the relationship to contain only the latest (highest) version of each A. Following the docs, I tried to solve this with a custom primaryjoin and a window function:

partition = select(
    A,
    row_number()
    .over(order_by=A.version.desc(), partition_by=A.id)
    .label("index"),
).alias()

partitioned_as = aliased(A, partition)

B.latest_as = relationship(
    partitioned_as,
    primaryjoin=and_(
        partition.c.index == 1,
        and_(
            partitioned_as.id == A2B.a_id,
            partitioned_as.version == A2B.a_version,
        ),
    ),
    secondary="a2b_table",
    viewonly=True,
)

Unfortunately, it doesn't work and I get the warning:

SELECT statement has a cartesian product between FROM element(s) "anon_1", "a2b_table" and FROM element "a_table". Apply join condition(s) between each element to resolve.

I checked the SQL statement sqlalchemy generates and it has anon_1, i.e. the query of partition, and a_table in its FROM clause. As far as I understand it, a_table shouldn't be in the FROM clause of this statement because it is already in the FROM clause of partition. I don't know how to get rid of it.

Could anyone point me in the right direction? Thanks in advance.

Recommended topics

Hot tags