SQLAlchemy - не применяет ограничение внешнего ключа для отношений

У меня есть модель/таблица 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 с entityId, который больше не подключается к какому-либо Test.id, потому что Test был удален. Однако, когда я пытаюсь создать TestAuditLog с SQLAlchemy, а затем удалю Test, я получаю сообщение об ошибке:

(IntegrityError) ( "23000", "[23000] [Microsoft] [Драйвер SQL-сервера ODBC] [SQL Server] Невозможно вставить значение NULL в столбец" AL_TEST_ID ", таблицу" TEST_AUDIT_LOG ", столбец не допускает нулей. UPDATE не работает. (515) (SQLExecDirectW); [01000] [Microsoft] [драйвер SQL-сервера ODBC] [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

Ответы

Ответ 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. Таким образом, вы можете проверить контрольный след удаленных объектов с помощью этого столбца.