SQLAlchemy - 不要強制執行關係的外鍵約束

Question

我有一个Test模型/表和一个TestAuditLog模型/表，使用SQLAlchemy和SQL Server 2008。两者之间的关系是Test.id == TestAuditLog.entityId，一个测试有许多审计日志。TestAuditLog旨在保留Test表中行的更改历史。我想跟踪Test何时被删除，但我对此有困难。在sqlservermanagementstudio中，我将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]Cannot insert the value NULL into column 'AL_TEST_ID', table 'TEST_AUDIT_LOG'; column does not allow nulls. UPDATE fails. (515) (SQLExecDirectW); [01000] [Microsoft][ODBC SQL Server Driver][SQL Server]The statement has been terminated. (3621)") u'UPDATE [TEST_AUDIT_LOG] SET [AL_TEST_ID]=? WHERE [TEST_AUDIT_LOG].[AL_ID] = ?' (None, 8)

我认为由于Test和TestAuditLog之间的外键关系，在我删除Test行之后，SQLAlchemy试图更新所有该测试的审计日志，使其具有NULLentityId。我不希望它这样做；我希望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

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

• 点1：在RDBMS级别和SA级别都没有ForeignKey
• 第2点：显式指定关系的连接条件
• 第3点：标记关系级联以依赖passive_deletes标志

下面完整工作的代码片段应该给您一个想法（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列可以为空。这样，您仍然可以使用此列检查已删除对象的审核跟踪。