Sqlite/SQLAlchemy:如何强制外键？

Question

新版本的SQLite能够强制执行外键约束，但是为了向后兼容，必须分别为每个数据库连接打开它！

sqlite> PRAGMA foreign_keys = ON;

我正在使用SQLAlchemy——我如何确保它始终处于打开状态？ 我试过的是：

engine = sqlalchemy.create_engine('sqlite:///:memory:', echo=True)
engine.execute('pragma foreign_keys=on')

……但它不起作用！。。。我错过了什么？

编辑： 我想我真正的问题是我安装了不止一个版本的SQLite，而Python没有使用最新的版本！

>>> import sqlite3
>>> print sqlite3.sqlite_version
3.3.4

但我刚刚下载了3.6.23并将exe放到我的项目目录中！ 我怎样才能知道它使用的是哪一个.exe并更改它？

Answer 1

根据conny和shadowmatter给出的答案，下面的代码将在发出PRAGMA语句之前检查您是否正在使用SQLite3：

from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlite3 import Connection as SQLite3Connection

@event.listens_for(Engine, "connect")
def _set_sqlite_pragma(dbapi_connection, connection_record):
    if isinstance(dbapi_connection, SQLite3Connection):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON;")
        cursor.close()

Answer 2

我现在有这个工作：

如上所述下载最新的sqlite和pysqlite2构建：确保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

下一步添加池侦听器：

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

那就小心你如何测试外键是否工作：我在这里有些困惑。当使用sqlalchemy ORM处理add()事物时，我的导入代码隐式地处理关系连接，因此永远不会失败。把nullable=False添加到一些ForeignKey()语句中有助于我实现这一点。

我测试sqlalchemy sqlite外键支持的方法是从声明性ORM类执行手动插入：

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

这里wall_id和type_id都是ForeignKey()，如果尝试连接无效的fkey，sqlite现在会正确抛出异常。所以有效！如果删除侦听器，那么sqlalchemy会很高兴地添加无效的条目。

我认为主要的问题可能是多个sqlite3.dll（或.so）存在。

Answer 3

对于最新版本（SQLAlchemy~0.7），SQLAlchemy homepage表示：

PoolListener is deprecated. Please refer to PoolEvents.

然后CarlS的例子变成：

engine = create_engine(database_url)

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)