Sqlite/Sqlalchemy: How to Enforce Foreign Keys

Sqlite / SQLAlchemy: how to enforce Foreign Keys?

I now have this working:

Download the latest sqlite and pysqlite2 builds as described above: make sure correct versions are being used at runtime by python.

import sqlite3   
import pysqlite2 
print sqlite3.sqlite_version   # should be 3.6.23.1
print pysqlite2.__path__       # eg C:\\Python26\\lib\\site-packages\\pysqlite2

Next add a PoolListener:

from sqlalchemy.interfaces import PoolListener
class ForeignKeysListener(PoolListener):
    def connect(self, dbapi_con, con_record):
        db_cursor = dbapi_con.execute('pragma foreign_keys=ON')

engine = create_engine(database_url, listeners=[ForeignKeysListener()])

Then be careful how you test if foreign keys are working: I had some confusion here. When using sqlalchemy ORM to add() things my import code was implicitly handling the relation hookups so could never fail. Adding nullable=False to some ForeignKey() statements helped me here.

The way I test sqlalchemy sqlite foreign key support is enabled is to do a manual insert from a declarative ORM class:

# example
ins = Coverage.__table__.insert().values(id = 99,
                                    description = 'Wrong',
                                    area = 42.0,
                                    wall_id = 99,  # invalid fkey id
                                    type_id = 99)  # invalid fkey_id
session.execute(ins) 

Here wall_id and type_id are both ForeignKey()'s and sqlite throws an exception correctly now if trying to hookup invalid fkeys. So it works! If you remove the listener then sqlalchemy will happily add invalid entries.

I believe the main problem may be multiple sqlite3.dll's (or .so) lying around.

How to correctly add Foreign Key constraints to SQLite DB using SQLAlchemy

SQLite by default does not enforce ForeignKey constraints (see here http://www.sqlite.org/pragma.html#pragma_foreign_keys )

To enable, follow these docs here: http://docs.sqlalchemy.org/en/latest/dialects/sqlite.html#foreign-key-support

Here's a copy paste of the official documentation:

SQLite supports FOREIGN KEY syntax when emitting CREATE statements for tables, however by default these constraints have no effect on the operation of the table.

Constraint checking on SQLite has three prerequisites:

• At least version 3.6.19 of SQLite must be in use
• The SQLite library must be compiled without the SQLITE_OMIT_FOREIGN_KEY or SQLITE_OMIT_TRIGGER symbols enabled.
• The PRAGMA foreign_keys = ON statement must be emitted on all connections before use.
  SQLAlchemy allows for the PRAGMA statement to be emitted automatically for new connections through the usage of events:

from sqlalchemy.engine import Engine
from sqlalchemy import event

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

How to turn on 'PRAGMA foreign_keys = ON' in sqlalchemy migration script or configuration file for sqlite?

See Foreign Key Support from SA SQLite documentation:

import sqlite3

from sqlalchemy.engine import Engine
from sqlalchemy import event

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    if type(dbapi_connection) is sqlite3.Connection:  # play well with other DB backends
       cursor = dbapi_connection.cursor()
       cursor.execute("PRAGMA foreign_keys=ON")
       cursor.close()

How can I enforce SQLite foreign keys in python using pysqlite2?

Try using SQLite's PRAGMA foreign_keys=ON; command, it is disabled by default.

Foreign key contraint is not getting set in SQLlite using flask alchemy

below code worked for me and found it from below link -

Sqlite / SQLAlchemy: how to enforce Foreign Keys?

   def _fk_pragma_on_connect(dbapi_con, con_record):
    dbapi_con.execute('pragma foreign_keys=ON')

from sqlalchemy import event
event.listen(engine, 'connect', _fk_pragma_on_connect)

Scrapy - SQLalchemy Foreign Key not created in SQLite

From the code I can see, it doesn't look to me like you are setting ThreadInfo.authorInfosLink or ThreadInfo.authorInfos_id anywhere (the same goes for all of your FK/relationships).

For the related objects to be attached to a ThreadInfo instance, you need to create them and then attach them something like:

        # Input info to authorInfo
        authorInfo = AuthorInfo()
        authorInfo.threadAuthor = item['threadAuthor'] 
        
        threadInfo.authorInfosLink = authorInfo

You probably don't want to session.add() each object if it's related via FK. You'll want to:

1. instantiate a BoardInfo object bi
2. then instantiate attach your related ThreadInfo object ti
3. attach your the related object eg bi.threadInfosLink = ti
4. At the end of all of your chained relationships, you can simply add bi to the session using session.add(bi) -- all of the related objects will be added through their relationships and the FKs will be correct.

SQLite PRAGMA foreign_keys with SQLAlchemy

we don't have an explicit API for event "removal" at this time though this is a feature that will eventually be available. So you'd need to create a single event that itself turns itself on and off based on a flag, seems like that's what you've already worked out.

Foreign key constraints in SQLAlchemy

It looks like the answer was in the database I was using (SQLite), not SQLAlchemy. SQLite versions <3.6.1 (AFAIK) do not support foreign key constraints.

The answer is therefore very similar to this answer on foreign keys and SQLAlchemy.

As I'm using Windows, I was able to go to the pysqlite2 page, the packaged installers have version 3.7.6.2 sqlite, and then and the final implementation was aided by this SQLAlchemy page on sqlite engines and dialects. This SO question is also relevant with regards to the upgrade process.

Finally, the SQLite engine is a bit temperamental when deciding whether or not to enforce the foreign key constraint, and this SO question is quite useful in forcing the foreign key enforcement.

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

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.

---

Related Topics