SQLAlchemy - 不强制关系上的外键约束

Question

我有一个叫做 Test 的模型/表和一个叫做 TestAuditLog 的模型/表，使用的是 SQLAlchemy 和 SQL Server 2008。这两个表之间的关系是 Test.id == TestAuditLog.entityId，也就是说一个测试可以有很多审计日志。TestAuditLog 主要是用来记录 Test 表中行的变化历史。我还想追踪什么时候删除了一个 Test，但是我在这方面遇到了一些问题。在 SQL Server Management Studio 中，我把 FK_TEST_AUDIT_LOG_TEST 关系的“强制外键约束”属性设置为“否”，我以为这样可以让 TestAuditLog 表中存在一个 entityId，即使这个 entityId 不再连接到任何 Test.id，因为那个 Test 已经被删除了。然而，当我尝试用 SQLAlchemy 创建一个 TestAuditLog，然后删除 Test 时，我收到了一个错误：

(IntegrityError) ('23000', "[23000] [Microsoft][ODBC SQL Server Driver][SQL Server]无法将 NULL 值插入到 'AL_TEST_ID' 列，表 'TEST_AUDIT_LOG'；该列不允许为空。更新失败。 (515) (SQLExecDirectW); [01000] [Microsoft][ODBC SQL Server Driver][SQL Server]语句已终止。 (3621)") u'UPDATE [TEST_AUDIT_LOG] SET [AL_TEST_ID]=? WHERE [TEST_AUDIT_LOG].[AL_ID] = ?' (None, 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

Answer 1

你可以通过以下方式来解决这个问题：

• 第一点：在数据库层面和SA层面都不要使用ForeignKey（外键）
• 第二点：明确指定关系的连接条件
• 第三点：标记关系的级联操作，依赖于passive_deletes标志

下面的完整代码示例应该能给你一些启发（要点在代码中突出显示）：

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

另一种选择是复制外键使用的列，并将外键列设置为可空，使用ON CASCADE SET NULL选项。这样，你仍然可以通过这个列来检查被删除对象的审计记录。