Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

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

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
like image 857
Sarah Vessels Avatar asked Aug 18 '11 15:08

Sarah Vessels


People also ask

What is Backref in SQLAlchemy?

The sqlalchemy backref is one of the type keywords and it passed as the separate argument parameters which has to be used in the ORM mapping objects. It mainly includes the event listener on the configuration attributes with both directions of the user datas through explicitly handling the database relationships.

What does relationship do in SQLAlchemy?

The relationship function is a part of Relationship API of SQLAlchemy ORM package. It provides a relationship between two mapped classes. This corresponds to a parent-child or associative table relationship.

Does SQLAlchemy require primary key?

The SQLAlchemy ORM, in order to map to a particular table, needs there to be at least one column denoted as a primary key column; multiple-column, i.e. composite, primary keys are of course entirely feasible as well.

Can a foreign key be null?

A table can have many foreign keys. A foreign key is nullable if any part is nullable. A foreign key value is null if any part is null.


1 Answers

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.

like image 72
van Avatar answered Sep 25 '22 03:09

van