SQLAlchemy - don't enforce foreign key constraint on a relationship

Question

I have a Test model/table and a TestAuditLog model/table, using SQLAlchemy and SQL Server 2008. The relationship between the two is Test.id == TestAuditLog.entityId, with one test having many audit logs. TestAuditLog is intended to keep a history of changes to rows in the Test table. I want to track when a Test is deleted, also, but I'm having trouble with this. In SQL Server Management Studio, I set the FK_TEST_AUDIT_LOG_TEST relationship's "Enforce Foreign Key Constraint" property to "No", thinking that would allow a TestAuditLog row to exist with an entityId that no longer connects to any Test.id because the Test has been deleted. However, when I try to create a TestAuditLog with SQLAlchemy and then delete the Test, I get an error:

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

I think because of the foreign-key relationship between Test and TestAuditLog, after I delete the Test row, SQLAlchemy is trying to update all that test's audit logs to have a NULL entityId. I don't want it to do this; I want SQLAlchemy to leave the audit logs alone. How can I tell SQLAlchemy to allow audit logs to exist whose entityId does not connect with any Test.id?

I tried just removing the ForeignKey from my tables, but I'd like to still be able to say myTest.audits and get all of a test's audit logs, and SQLAlchemy complained about not knowing how to join Test and TestAuditLog. When I then specified a primaryjoin on the relationship, it grumbled about not having a ForeignKey or ForeignKeyConstraint with the columns.

Here are my models:

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

And here's how I'm trying to delete a test while keeping its audit logs, their entityId intact:

    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

You can solve this by:

• POINT-1: not having a ForeignKey neither on the RDBMS level nor on the SA level
• POINT-2: explicitly specify join conditions for the relationship
• POINT-3: mark relationship cascades to rely on passive_deletes flag

Fully working code snippet below should give you an idea (points are highlighted in the 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

Another option would be to duplicate the column used in the FK, and make the FK column nullable with ON CASCADE SET NULL option. In this way you can still check the audit trail of deleted objects using this column.