在sqlalchemy中创建数据库时,如何摆脱循环依赖错误？

Question

我是使用sqlalchemy的新手.如何摆脱下面显示的表的循环依赖性错误.基本上我的目标是创建一个问题表,其中一对一关系"最佳答案"可以回答,一对多关系也可以"possible_answers".

class Answer(Base):
    __tablename__ = 'answers'
    id = Column(Integer, primary_key=True)
    text = Column(String)

    question_id = Column(Integer, ForeignKey('questions.id'))

    def __init__(self, text, question_id):
        self.text = text

    def __repr__(self):
        return "<Answer '%s'>" % self.text

class Question(Base):
    __tablename__ = 'questions'

    id = Column(Integer, primary_key=True)
    text = Column(String)
    picture = Column(String)
    depth = Column(Integer)
    amount_of_tasks = Column(Integer)
    voting_threshold = Column(Integer)
    best_answer_id = Column(Integer, ForeignKey('answers.id'), nullable=True)

    possible_answers = relationship("Answer", post_update=True, primaryjoin = id==Answer.question_id)

    def __init__(self, text, picture, depth, amount_of_tasks):
        self.text = text
        self.picture = picture
        self.depth = depth
        self.amount_of_tasks = amount_of_tasks

    def __repr__(self):
        return "<Question, '%s', '%s', '%s', '%s'>" % (self.text, self.picture, self.depth, self.amount_of_tasks)

    def __repr__(self):
        return "<Answer '%s'>" % self.text

这是错误消息:CircularDependencyError:检测到循环依赖性.周期:

Answer 1

马克的解决方案有效，但我想找到一种无需创建额外表即可实现的方法。经过大量搜索，我终于在文档中找到了这个示例：

http://docs.sqlalchemy.org/en/latest/orm/relationship_persistence.html（第二个示例）

该方法是对模型中的两种关系使用primaryjoin[1] Question，并添加post_update=True其中之一。告诉post_updatesqlalchemy 设置best_answer_id为附加UPDATE语句，绕过循环依赖。

您还需要foreign_keys指定模型question中的关系Answer。

下面是经过修改的 Mark 代码，以遵循上面链接的示例。我用 sqlalchemy 测试了它v1.1.9。

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

class Answer(Base):
    __tablename__ = 'answer'
    id = Column(Integer, primary_key=True)
    text = Column(String)
    question_id = Column(Integer, ForeignKey('question.id'))
    question = relationship('Question', back_populates='answers', foreign_keys=[question_id])

    def __repr__(self):
        return "<Answer '%s'>" % self.text

class Question(Base):
    __tablename__ = 'question'
    id = Column(Integer, primary_key=True)
    text = Column(String)
    best_answer_id = Column(Integer, ForeignKey('answer.id'))
    answers     = relationship('Answer', primaryjoin= id==Answer.question_id)
    best_answer = relationship('Answer', primaryjoin= best_answer_id==Answer.id, post_update=True)

    def __repr__(self):
        return "<Question, '%s'>" % (self.text)

if __name__ == '__main__':

    session = sessionmaker(bind=engine)()
    Base.metadata.create_all(engine)

    question = Question(text='How good is SQLAlchemy?')

    somewhat = Answer(text='Somewhat good')
    very = Answer(text='Very good')
    excellent = Answer(text='Excellent!')

    question.answers.extend([somewhat, very, excellent])
    question.best_answer = excellent

    session.add(question)
    session.commit()

    question = session.query(Question).first()

    print(question.answers)
    print(question.best_answer)

[1] 有趣的是，“字符串格式”primaryjoin似乎会导致错误——但是使用列对象上的重载运算符构建 SQL 表达式是可行的。

Answer 2

显然,SQLAlchemy不能很好地处理循环依赖.您可以考虑使用关联表来代表最佳答案......

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy import Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()


class Answer(Base):
    __tablename__ = 'answer'

    id = Column(Integer, primary_key=True)
    question_id = Column(Integer, ForeignKey('question.id'))
    text = Column(String)

    question = relationship('Question', backref='answers')

    def __repr__(self):
        return "<Answer '%s'>" % self.text


class Question(Base):
    __tablename__ = 'question'

    id = Column(Integer, primary_key=True)
    text = Column(String)

    best_answer = relationship('Answer',
                               secondary=lambda: best_answer,
                               uselist=False)

    def __repr__(self):
        return "<Question, '%s'>" % (self.text)

best_answer = Table('best_answer', Base.metadata,
                    Column('question_id',
                           Integer,
                           ForeignKey('question.id'),
                           primary_key=True),
                    Column('answer_id',
                           Integer,
                           ForeignKey('answer.id'))
                    )


if __name__ == '__main__':

    session = sessionmaker(bind=engine)()
    Base.metadata.create_all(engine)

    question = Question(text='How good is SQLAlchemy?')

    somewhat = Answer(text='Somewhat good')
    very = Answer(text='Very good')
    excellent = Answer(text='Excellent!')

    question.answers.extend([somewhat, very, excellent])
    question.best_answer = excellent

    session.add(question)
    session.commit()

    question = session.query(Question).first()

    print(question.answers)
    print(question.best_answer)