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不再连接到任何行的行,Test.id因为Test已删除.但是,当我尝试TestAuditLog使用SQLAlchemy 创建一个然后删除它时Test,我收到一个错误:

(IntegrityError)('23000',"[23000] [Microsoft] [ODBC SQL Server驱动程序] [SQL Server]无法将值NULL插入列'AL_TEST_ID',表'TEST_AUDIT_LOG';列不允许空值.UPDATE失败. (515)(SQLExecDirectW); [01000] [Microsoft] [ODBC SQL Server驱动程序] [SQL Server]语句已终止.(3621)")u'UPDATE [TEST_AUDIT_LOG] SET [AL_TEST_ID] =？WHERE [TEST_AUDIT_LOG].[AL_ID] =？' (无,8)

我想是因为之间的外键关系Test和TestAuditLog,我删除后Test列,SQLAlchemy的试图更新所有测试的审计日志有NULL entityId.我不希望它这样做; 我希望SQLAlchemy单独保留审计日志.如何告诉SQLAlchemy允许存在entityId不与任何连接的审计日志Test.id？

我试图ForeignKey从我的表中删除,但我仍然可以说myTest.audits并获得所有测试的审计日志,并且SQLAlchemy抱怨不知道如何加入Test和TestAuditLog.当我再指定primaryjoin的relationship,它抱怨不具有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

你可以通过以下方式解决

• POINT-1:不具有ForeignKey既不在RDBMS水平或在水平SA
• POINT-2:明确指定关系的连接条件
• POINT-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中使用的列,并使FK列可以ON CASCADE SET NULL选择为空.通过这种方式,您仍然可以使用此列检查已删除对象的审计跟踪.