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

对于最新版本 (SQLAlchemy ~0.7)，SQLAlchemy homepage 表示：

PoolListener 已弃用。请参考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)

Answer 2

基于 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 3

我现在有这个工作：

如上所述下载最新的 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

接下来添加一个 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()])

然后小心你如何测试外键是否有效：我在这里有些困惑。当使用 sqlalchemy ORM 处理add() 的事情时，我的导入代码会隐式处理关系连接，因此永远不会失败。在一些 ForeignKey() 语句中添加 nullable=False 对我有帮助。

我测试启用 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 4

来自SQLite dialect page：

SQLite 在为表发出 CREATE 语句时支持 FOREIGN KEY 语法，但默认情况下，这些约束对表的操作没有影响。

对 SQLite 的约束检查具有三个先决条件：

• 必须至少使用 3.6.19 版的 SQLite
• 必须在未启用 SQLITE_OMIT_FOREIGN_KEY 或 SQLITE_OMIT_TRIGGER 符号的情况下编译 SQLite 库。
• 使用前必须在所有连接上发出 PRAGMA foreign_keys = ON 语句。

SQLAlchemy 允许通过使用事件为新连接自动发出 PRAGMA 语句：

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

Answer 5

如果您的会话创建集中在 Python 辅助函数后面（而不是直接公开 SQLA 引擎），作为一种更简单的方法，您可以在返回新创建的会话之前发出 session.execute('pragma foreign_keys=on')。

如果您的应用程序的任意部分可能会针对数据库创建 SQLA 会话，则您只需要池侦听器方法。

Answer 6

我之前也遇到过同样的问题（带有外键约束的脚本正在执行，但 sqlite 引擎并未强制执行实际约束）；解决了：

1. 从这里下载、构建和安装最新版本的 sqlite：sqlite-sqlite-amalgamation;在此之前，我的 ubuntu 机器上有 sqlite 3.6.16；尚不支持外键；应该是 3.6.19 或更高版本才能让它们工作。

2. 从这里安装最新版本的 pysqlite：pysqlite-2.6.0

之后，每当外键约束失败时，我都会收到异常

希望这会有所帮助，问候

Answer 7

如果您需要在每个连接上执行设置，请使用PoolListener。

Answer 8

单行版本的conny's answer：

from sqlalchemy import event

event.listen(engine, 'connect', lambda c, _: c.execute('pragma foreign_keys=on'))

Answer 9

在使用 Flask + SQLAlchemy 时对 sqlite 实施外键约束。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

def create_app(config: str=None):
    app = Flask(__name__, instance_relative_config=True)
    if config is None:
        app.config.from_pyfile('dev.py')
    else:
        logger.debug('Using %s as configuration', config)
        app.config.from_pyfile(config)

    db.init_app(app)

    # Ensure FOREIGN KEY for sqlite3
    if 'sqlite' in app.config['SQLALCHEMY_DATABASE_URI']:
        def _fk_pragma_on_connect(dbapi_con, con_record):  # noqa
            dbapi_con.execute('pragma foreign_keys=ON')

        with app.app_context():
            from sqlalchemy import event
            event.listen(db.engine, 'connect', _fk_pragma_on_connect)

来源： https://gist.github.com/asyd/a7aadcf07a66035ac15d284aef10d458