SQLalchemy 相互依赖的外键约束

Question

我正在尝试定义 2 个这样的实体：

class User(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(256), index=True, unique=True)
    main_token_id = Column(ForeignKey('token.id'), nullable=False)

    main_token = relationship('Token', uselist=False)
    tokens = relationship('Token', back_populates="user", foreign_keys=['token.id'])


class Token(Base):
    id = Column(Integer, primary_key=True)
    user_id = Column(ForeignKey('user.id'), nullable=False)

    user: User = relationship("user", back_populates="tokens")

我希望用户能够访问他所有令牌的集合，并且我还希望他拥有一个特殊的主令牌。我想确保用户只有一个主令牌，并且我需要外键提供的完整性。实际上是他们俩。

我已经阅读了Cascading deletes in mutually dependent tables in SQLAlchemy，但我觉得它没有帮助。我想拥有双方的诚信。

我怎样才能做到这一点？如果设计有缺陷，我该如何重新表述，以保持我的完整性保证？

Answer 1

我以前用来解决这个问题的方法是在令牌上创建一个类似precedence = Column(Integer, nullable=False) 的列。然后设置一个唯一约束，如UniqueConstraint('user_id', 'precedence')。然后在创建令牌时手动设置该整数。优先级为 0 或最低优先级的令牌是 主令牌。

这是一个例子。我确信一些 sqlalchemy 天才可以在没有 3 次更新的情况下执行优先级交换，但我认为在大多数情况下不会经常出现。有一种方法可以延迟事务中的唯一约束，但我猜 sqlite 还不支持。

这依赖于您的应用程序不会从优先级 0 中清除主令牌，即。没有完整性检查来防止这种情况发生。

from sqlalchemy import (
    create_engine,
    UnicodeText,
    Integer,
    String,
    ForeignKey,
    UniqueConstraint,
    update,
)
from sqlalchemy.schema import (
    Table,
    Column,
    MetaData,
)
from sqlalchemy.sql import select
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError


Base = declarative_base()


engine = create_engine("sqlite://", echo=False)


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(256), index=True, unique=True)
    tokens = relationship('Token', backref="user", cascade="all, delete-orphan", order_by='Token.precedence')

    main_token = relationship('Token', primaryjoin='and_(User.id == Token.user_id, Token.precedence == 0)', viewonly=True, uselist=False)


class Token(Base):
    __tablename__ = 'tokens'
    id = Column(Integer, primary_key=True)
    precedence = Column(Integer, nullable=False)
    user_id = Column(ForeignKey('users.id'), nullable=False)
    __table_args__ = (UniqueConstraint('precedence', 'user_id', name='tokens_user_precedence'),)


Base.metadata.create_all(engine)


with Session(engine) as session:
    user = User(name='tokenizer')
    session.add(user)
    main_token = Token(user=user, precedence=0)
    session.add(main_token)
    session.add(Token(user=user, precedence=1))
    session.commit()
    assert session.query(Token).first()
    assert session.query(User).first()
    assert session.query(User).first().tokens
    assert session.query(User).first().tokens[0] == main_token
    # This viewonly relationship seems to be working.
    assert session.query(User).first().main_token == main_token

    # We don't want this so don't do this, no integrity checks here!!
    main_token.precedence = 100
    session.commit()
    assert not session.query(User).first().main_token

    # Put it back now.
    main_token.precedence = 0
    session.commit()
    assert session.query(User).first().main_token

    # Now check tokens are cleared.
    session.delete(user)
    session.commit()
    assert not session.query(Token).all()
    assert not session.query(User).all()


with Session(engine) as session:
    # Try making 2 main tokens.
    user = User(name='tokenizer')
    session.add(user)
    main_token = Token(user=user, precedence=0)
    main_token2 = Token(user=user, precedence=0)
    session.add_all([main_token, main_token2])
    try:
        session.commit()
    except IntegrityError as e:
        pass
    else:
        assert False, 'Exception should have occurred.'

with Session(engine) as session:
    # Try swapping the tokens.
    user = User(name='tokenizer')
    session.add(user)
    main_token = Token(user=user, precedence=0)
    session.add(main_token)
    other_token = Token(user=user, precedence=1)
    session.add(other_token)
    session.commit()
    old_precedence = other_token.precedence
    main_token.precedence = -1
    session.flush()
    other_token.precedence = 0
    session.flush()
    main_token.precedence = old_precedence
    session.commit()
    user.tokens[0] == other_token
    user.tokens[1] == main_token
    user.main_token == other_token
    session.commit()