SQLAlchemy：对同一个表和附加数据的多对多关系感到困惑

Question

我希望用户成为推荐人和推荐人，并存储一些额外的数据。这是我的代码：

class User(UserMixin, db.Model):
    __tablename__ = 'user'
    referees = db.relationship('RefData', back_populates='referral', lazy='dynamic')
    referrals = db.relationship('RefData', back_populates='referee', lazy='dynamic')

和参考数据表：

class RefData(UserMixin, db.Model):
    __tablename__ = 'ref_data'

    referral_id = db.Column(UUID(as_uuid=True), db.ForeignKey('user.id'), primary_key=True)
    referral = db.relationship('User', back_populates='referrals')

    referee_id = db.Column(UUID(as_uuid=True), db.ForeignKey('user.id'), primary_key=True)
    referee = db.relationship('User', back_populates='referees')

但是使用这段代码，我遇到了以下异常，我不知道如何解决它：

sqlalchemy.exc.AmbiguousForeignKeysError: Could not determine join condition between 
parent/child tables on relationship User.referees - there are multiple foreign key paths 
linking the tables.  Specify the 'foreign_keys' argument, providing a list of those columns 
which should be counted as containing a foreign key reference to the parent table.

Answer 1

对于 SQLAlchemy 2.0：

from __future__ import annotations

from sqlalchemy import ForeignKey, create_engine
from sqlalchemy.orm import (DeclarativeBase, Mapped, MappedAsDataclass,
                            mapped_column, relationship, sessionmaker)

DATABASE_URL = "sqlite+pysqlite:///store.db"

engine = create_engine(
    DATABASE_URL,
    # echo=True,
)

SessionMaker = sessionmaker(
    bind=engine,
)


class Base(MappedAsDataclass, DeclarativeBase):
    pass


class User(Base):
    __tablename__ = 'user'

    user_id: Mapped[int] = mapped_column(
        primary_key=True,
        init=False,
    )

    name: Mapped[str] = mapped_column(
        default="",
    )

    referee_associations: Mapped[list[Association]] = relationship(
        primaryjoin="User.user_id == Association.referrer_id",
        back_populates="referrer",
        init=False,
        repr=False,
    )

    referrer_associations: Mapped[list[Association]] = relationship(
        primaryjoin="User.user_id == Association.referee_id",
        back_populates="referee",
        init=False,
        repr=False,
    )


class Association(Base):
    __tablename__ = 'association'

    referrer_id: Mapped[int] = mapped_column(
        ForeignKey("user.user_id"),
        primary_key=True,
        default=None,
    )

    referee_id: Mapped[int] = mapped_column(
        ForeignKey("user.user_id"),
        primary_key=True,
        default=None,
    )

    data: Mapped[str] = mapped_column(
        default="",
    )

    referrer: Mapped[User] = relationship(
        back_populates="referee_associations",
        foreign_keys=[referrer_id],
        # init=False,
        repr=False,
        default=None,
    )

    referee: Mapped[User] = relationship(
        back_populates="referrer_associations",
        foreign_keys=[referee_id],
        # init=False,
        repr=False,
        default=None,
    )


if __name__ == "__main__":
    Base.metadata.create_all(engine)

    with SessionMaker() as session:
        # create some users
        a = User(name="Alice")
        b = User(name="Bob")
        c = User(name="Charlie")
        d = User(name="David")
        e = User(name="Eve")
        session.add_all([a, b, c, d, e])
        session.commit()

        # let's say A, B  are referrers of C, D and E
        a1 = Association(referrer=a, referee=c, data="a->c")
        a2 = Association(referrer=a, referee=d, data="a->d")
        a3 = Association(referrer=a, referee=e, data="a->e")
        a4 = Association(referrer=b, referee=c, data="b->c")
        a5 = Association(referrer=b, referee=d, data="b->d")
        a6 = Association(referrer=b, referee=e, data="b->e")
        session.add_all([a1, a2, a3, a4, a5, a6])
        session.commit()

        # check if it works
        alice = session.get(User, a.user_id)
        if alice is not None:
            print(f"# Referee for {alice.name}:")
            if alice.referee_associations is not None:
                for r in alice.referee_associations:
                    print(f"{r.referee.name:<10}", r.data)

        david = session.get(User, d.user_id)
        if david is not None:
            print(f"# Referrer for {david.name}:")
            if david.referrer_associations is not None:
                for r in david.referrer_associations:
                    print(f"{r.referrer.name:<10}", r.data)

执行上面的命令给我们：

# Referee for Alice:
Charlie    a->c
David      a->d
Eve        a->e
# Referrer for David:
Alice      a->d
Bob        b->d