SQLAlchemy - không thực thi ràng buộc khóa ngoài đối với mối quan hệ

Question

Tôi có mô hình/bảng Test và mô hình TestAuditLog/bảng, sử dụng SQLAlchemy và SQL Server 2008. Mối quan hệ giữa hai là Test.id == TestAuditLog.entityId, với một thử nghiệm có nhiều kiểm tra nhật ký. TestAuditLog nhằm giữ lịch sử thay đổi đối với các hàng trong bảng Test. Tôi cũng muốn theo dõi thời điểm Test cũng bị xóa nhưng tôi đang gặp sự cố với điều này. Trong SQL Server Management Studio, tôi đặt thuộc tính "Enforce Foreign Key Constraint" của mối quan hệ là "" thành "Không", cho phép hàng TestAuditLog tồn tại với entityId không còn kết nối với bất kỳ Test.id nào vì Test đã bị xóa. Tuy nhiên, khi tôi cố gắng để tạo ra một TestAuditLog với SQLAlchemy và sau đó xóa các Test, tôi nhận được một lỗi:

(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)

Tôi nghĩ rằng vì mối quan hệ ngoại-key giữa Test và TestAuditLog, sau khi tôi xóa hàng Test, SQLAlchemy đang cố gắng cập nhật tất cả nhật ký kiểm tra của thử nghiệm để có một NULLentityId. Tôi không muốn nó làm điều này; Tôi muốn SQLAlchemy để lại các bản ghi kiểm toán một mình. Làm cách nào tôi có thể yêu cầu SQLAlchemy cho phép nhật ký kiểm tra tồn tại có entityId không kết nối với bất kỳ Test.id nào?

tôi đã cố gắng chỉ loại bỏ các ForeignKey từ các bảng của tôi, nhưng tôi muốn vẫn có thể nói myTest.audits và nhận được tất cả các bản ghi kiểm toán của kiểm tra, và SQLAlchemy phàn nàn về việc không biết làm thế nào để tham gia Test và TestAuditLog. Khi tôi sau đó chỉ định một primaryjoin trên relationship, nó càu nhàu về việc không có ForeignKey hoặc ForeignKeyConstraint với các cột.

Dưới đây là mô hình của tôi:

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") 
    ... 

Và dưới đây là cách tôi đang cố gắng để xóa một bài kiểm tra trong khi vẫn giữ các bản ghi kiểm toán, họ entityId nguyên vẹn:

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

Bạn có thể giải quyết điều này bằng cách:

• POINT-1: không có ForeignKey không phải trên mức RDBMS cũng vào mức độ SA
• POINT-2: định rõ điều kiện tham gia cho mối quan hệ
• POINT-3: thác mối quan hệ hiệu dựa vào passive_deletes cờ

Đoạn mã hoạt động hoàn toàn bên dưới sẽ cung cấp cho bạn ý tưởng (điểm được đánh dấu trong 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 

Một tùy chọn khác sẽ trùng lặp cột được sử dụng trong FK và làm cho cột FK có thể vô hiệu hóa với tùy chọn ON CASCADE SET NULL. Bằng cách này bạn vẫn có thể kiểm tra đường mòn kiểm toán của các đối tượng đã xóa bằng cách sử dụng cột này.