SQLAlchemy - 不要对关系强制外键约束

Posted 2023-03-16

【中文标题】SQLAlchemy - 不要对关系强制外键约束

【英文标题】：SQLAlchemy - don't enforce foreign key constraint on a relationship

【发布时间】：2011-10-29 22:06:50

【问题描述】：

我有一个Test 模型/表和一个TestAuditLog 模型/表，使用SQLAlchemy 和SQL Server 2008。两者之间的关系是Test.id == TestAuditLog.entityId，其中一个测试有许多审计日志。 TestAuditLog 旨在保留Test 表中行的更改历史记录。我也想跟踪Test 何时被删除，但我遇到了麻烦。在 SQL Server Management Studio 中，我将FK_TEST_AUDIT_LOG_TEST 关系的“Enforce Foreign Key Constraint”属性设置为“否”，认为这将允许TestAuditLog 行与不再连接到任何Test.id 的entityId 存在，因为Test 已被删除。但是，当我尝试使用 SQLAlchemy 创建 TestAuditLog 然后删除 Test 时，出现错误：

(IntegrityError) ('23000', "[23000] [Microsoft][ODBC SQL Server Driver][SQL Server]无法将值 NULL 插入到列 'AL_TEST_ID'、表 'TEST_AUDIT_LOG'；列不允许空值。 UPDATE 失败。(515) (SQLExecDirectW); [01000] [Microsoft][ODBC SQL Server Driver][SQL Server]语句已终止。(3621)") u'UPDATE [TEST_AUDIT_LOG] SET [AL_TEST_ID]=?在哪里 [TEST_AUDIT_LOG].[AL_ID] = ？ （无，8）

我认为由于Test 和TestAuditLog 之间的外键关系，在我删除Test 行后，SQLAlchemy 正在尝试更新所有该测试的审核日志以具有NULL entityId。我不希望它这样做；我希望 SQLAlchemy 不理会审计日志。如何告诉 SQLAlchemy 允许存在 entityId 不与任何 Test.id 连接的审计日志？

我尝试从我的表中删除 ForeignKey，但我仍然希望能够说 myTest.audits 并获取所有测试的审核日志，并且 SQLAlchemy 抱怨不知道如何加入 Test和TestAuditLog。当我在relationship 上指定primaryjoin 时，它抱怨没有ForeignKey 或ForeignKeyConstraint 与列。

这是我的模型：

class TestAuditLog(Base, Common):
    __tablename__ = u'TEST_AUDIT_LOG'
    entityId = Column(u'AL_TEST_ID', INTEGER(), ForeignKey(u'TEST.TS_TEST_ID'),
        nullable=False)
    ...

class Test(Base, Common):
    __tablename__ = u'TEST'
    id = Column(u'TS_TEST_ID', INTEGER(), primary_key=True, nullable=False)
    audits = relationship(TestAuditLog, backref="test")
    ...

这就是我尝试删除测试同时保留其审计日志的方式，他们的entityId 完好无损：

    test = Session.query(Test).first()
    Session.begin()
    try:
        Session.add(TestAuditLog(entityId=test.id))
        Session.flush()
        Session.delete(test)
        Session.commit()
    except:
        Session.rollback()
        raise

【参考方案1】：

您可以通过以下方式解决此问题：

要点 1：

在 RDBMS 级别和 SA 级别都没有 ForeignKey

POINT-2：

明确指定关系的连接条件

POINT-3：

标记关系级联以依赖passive_deletes 标志

下面的完整工作代码 sn-p 应该会给您一个想法（点在code 中突出显示）：

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

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

Session = sessionmaker(bind=engine)

class TestAuditLog(Base):
    __tablename__ = 'TEST_AUDIT_LOG'
    id = Column(Integer, primary_key=True)
    comment = Column(String)

    entityId = Column('TEST_AUDIT_LOG', Integer, nullable=False,
                     # POINT-1
                     #ForeignKey('TEST.TS_TEST_ID', ondelete="CASCADE"),
                     )

    def __init__(self, comment):
        self.comment = comment

    def __repr__(self):
        return "<TestAuditLog(id=%s entityId=%s, comment=%s)>" % (self.id, self.entityId, self.comment)

class Test(Base):
    __tablename__ = 'TEST'
    id = Column('TS_TEST_ID', Integer, primary_key=True)
    name = Column(String)

    audits = relationship(TestAuditLog, backref='test',
                # POINT-2
                primaryjoin="Test.id==TestAuditLog.entityId",
                foreign_keys=[TestAuditLog.__table__.c.TEST_AUDIT_LOG],
                # POINT-3
                passive_deletes='all',
            )

    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return "<Test(id=%s, name=%s)>" % (self.id, self.name)


Base.metadata.create_all(engine)

###################
## tests
session = Session()

# create test data
tests = [Test("test-" + str(i)) for i in range(3)]
_cnt = 0
for _t in tests:
    for __ in range(2):
        _t.audits.append(TestAuditLog("comment-" + str(_cnt)))
        _cnt += 1
session.add_all(tests)
session.commit()
session.expunge_all()
print '-'*80

# check test data, delete one Test
t1 = session.query(Test).get(1)
print "t: ", t1
print "t.a: ", t1.audits
session.delete(t1)
session.commit()
session.expunge_all()
print '-'*80

# check that audits are still in the DB for deleted Test
t1 = session.query(Test).get(1)
assert t1 is None
_q = session.query(TestAuditLog).filter(TestAuditLog.entityId == 1)
_r = _q.all()
assert len(_r) == 2
for _a in _r:
    print _a

另一个选项是复制 FK 中使用的列，并使用 ON CASCADE SET NULL 选项使 FK 列可以为空。这样您仍然可以使用此列检查已删除对象的审计跟踪。

【讨论】：

passive_deletes='all' 上的relationship 做到了！这样我就能够保持关系并且 SQLAlchemy 没有返回并试图消除 Test 删除上的 entityId。谢谢！

仅供参考 - 还需要在关系的父方设置lazy="dynamic"，因此当您不需要它时（即仅更新不相关的字段时，sqlalchemy 不会获取所有子项）在父表中）。

@Greg0ry：不，你不需要。正如Using Loader Strategies: Lazy Loading, Eager Loading 中所述：默认情况下，所有对象间关系都是延迟加载...。因此，除非您不这样做，否则除非您访问它们，否则父级不应加载子级。

@van - 在 Pylons Pyramid 中使用 sqlalchemy 时，我的观察结果有所不同 - 只有明确的 lasy="dynamic" 关系才能阻止 sqlalchemy 获取正在更新的父级的子级。

有趣，也许您的代码出于某种原因访问了孩子？数数或打印出来，还是其他原因？