我能让SQLAlchemy根据当前外键值填充关系吗?

11 投票
1 回答
5699 浏览
提问于 2025-04-18 03:19

这里有一段代码:

# latest version at https://gist.github.com/nickretallack/11059102

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

Base = declarative_base()

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False, unique=True)

    def __str__(self):
        return self.name

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(ForeignKey(Parent.id), nullable=False)
    name = Column(String, nullable=False)

    parent = relationship(Parent)

engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)

def run():
    # Basic Setup
    Base.metadata.create_all(engine)
    session = Session()
    fred = Parent(name="Fred", id=1)
    george = Parent(name="George", id=2)
    session.add(fred, george)
    session.commit()

    # The test
    bob = Child(name="Bob", parent_id=1)
    print bob.parent, ": Out of session. Should be Fred but is None.\n"

    session.add(bob)
    print bob.parent, ": In session.  Should be Fred but is None.\n"

    session.commit()
    print bob.parent, ": Committed.  Is Fred.\n" 

    bob.parent_id = 2
    print bob.parent, ": Dirty.  Should be George but is Fred.\n"

    session.add(bob)
    print bob.parent, ": Added to session.  Should be George but is Fred.\n"

    session.expire(bob,['parent'])
    print bob.parent, ": Expired.  Should be George but is None?  Wtf?\n"

    session.commit()
    print bob.parent, ": Committed again.  Is None.  Ugh.\n"

if __name__ == '__main__':
    run()

这个例子说明,仅仅设置一个关系所依赖的外键字段是不够的,无法让这个关系查询到正确的内容。无论我怎么做,这种情况几乎总是发生。

有没有办法让sqlalchemy根据当前的外键值来填充这个关系,而不需要先保存记录?我能做些什么来让它执行查询呢?

这个问题在处理网页表单时经常出现。表单提交的内容通常只包含一些ID,所以处理这些提交的最简单方法就是在记录中设置ID字段,然后尝试提交。如果引用的项目不存在,或者有其他只有数据库才能知道的问题(比如唯一性约束冲突),那么交易就会失败。一旦交易失败,你可能想要重新显示表单给用户。不幸的是,这时候所有的关系都不再正确了。

这可能是个问题,也可能不是,但在我的情况下,这确实让人很沮丧。为了修正这些关系,我需要重复这些关系中的逻辑,因为我找不到方法让它们直接执行查询。

1 个回答

18
  1. 你的“add”调用是错误的:

     session.add_all([fred, george])
    
  2. 对于一个完全临时的对象,它甚至不在会话中(顺便说一下,我不同意这种用法),可以使用 enable_relationship_loading

     # The test
     bob = Child(name="Bob", parent_id=1)
     session.enable_relationship_loading(bob)
     print bob.parent, ": Out of session. Should be Fred but is None.\n"
    
  3. 如果你想让一个待处理的对象加载它的关系(我也不同意这种用法,具体可以看 我把“foo_id”属性设置为“7”,但“foo”属性仍然是None - 难道它不应该加载id为7的Foo吗?),可以使用 load_on_pending 标志:

     class Child(Base):
         __tablename__ = 'child'
         id = Column(Integer, primary_key=True)
         parent_id = Column(ForeignKey(Parent.id), nullable=False)
         name = Column(String, nullable=False)
    
         parent = relationship(Parent, load_on_pending=True)
    
  4. 当你改变了“parent_id”时,如果想重新加载“parent”,可以使用expire,正如常见问题解答中提到的那样:

     session.expire(bob, ['parent'])
     bob.parent_id = 2
     print bob.parent, ": Dirty.  Should be George but is Fred.\n"
    

脚本完全正常工作:

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

Base = declarative_base()

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False, unique=True)

    def __str__(self):
        return self.name

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(ForeignKey(Parent.id), nullable=False)
    name = Column(String, nullable=False)

    parent = relationship(Parent, load_on_pending=True)

engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)

def run():
    # Basic Setup
    Base.metadata.create_all(engine)
    session = Session()
    fred = Parent(name="Fred", id=1)
    george = Parent(name="George", id=2)
    session.add_all([fred, george])
    session.commit()

    # The test
    bob = Child(name="Bob", parent_id=1)
    session.enable_relationship_loading(bob)
    print bob.parent, ": Out of session. Should be Fred but is None.\n"

    session.add(bob)
    print bob.parent, ": In session.  Should be Fred but is None.\n"

    session.commit()
    print bob.parent, ": Committed.  Is Fred.\n"

    session.expire(bob, ['parent'])
    bob.parent_id = 2
    print bob.parent, ": Dirty.  Should be George but is Fred.\n"

    session.add(bob)
    print bob.parent, ": Added to session.  Should be George but is Fred.\n"

    session.expire(bob,['parent'])
    print bob.parent, ": Expired.  Should be George but is None?  Wtf?\n"

    session.commit()
    print bob.parent, ": Committed again.  Is None.  Ugh.\n"

if __name__ == '__main__':
    run()

撰写回答