LoginSignup
6
2

More than 3 years have passed since last update.

Flask-SQLAlchemyでSQLiteを使用したときに外部キー制約を順守させる方法

Last updated at Posted at 2020-01-31

 外部キー制約に反した時にエラーを吐くようにする方法を探していた。
 結論:(見当たら)ない。諦めてコード側で守る。
    接続するたびにPRAGMA foreign_keysをONにする。コード参照。
 @tiechel様ありがとうございます。

検証方法

環境

 Python 3.7 3.8
 Flask-SQLAlchemy v2.4.1 v2.4.4
 

失敗例

2つのテーブルを作成する。

Model.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./tmp/test.db'
db = SQLAlchemy(app)

class Person(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    addresses = db.relationship('Address', backref='person', lazy=True)


class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), nullable=False)
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'),
                                nullable=False)

 https://flask-sqlalchemy.palletsprojects.com/en/2.x/quickstart/
 https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/ より

出来上がるテーブル

test.db
CREATE TABLE "person" (
    "id"    INTEGER NOT NULL,
    "name"  VARCHAR(50) NOT NULL,
    PRIMARY KEY("id")
);

CREATE TABLE "address" (
    "id"    INTEGER NOT NULL,
    "email" VARCHAR(120) NOT NULL,
    "person_id" INTEGER NOT NULL,
    PRIMARY KEY("id"),
    FOREIGN KEY("person_id") REFERENCES "person"("id")
);

テストを行う

test.py
import unittest

def testForeignKey(self):
    db.create_all()
    db.session.add(db.Address(id=0, email="aaa", person_id=111))
    db.session.commit()


if __name__ == '__main__':
    unittest.main()

 test.dbを覗いて新しいレコードが追加されていることを確認する。
 

外部キー制約に反した時にエラーを吐くようにする方法

Model.pyを更新する

Model.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# ---------変更点---------
import sqlalchemy.engine
from sqlalchemy import event
# ---------変更点---------

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./tmp/test.db'
db = SQLAlchemy(app)


# ---------変更点---------
@event.listens_for(sqlalchemy.engine.Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys = ON")
    cursor.close()
# ---------変更点---------

class Person(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    addresses = db.relationship('Address', backref='person', lazy=True)


class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), nullable=False)
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'),
                                nullable=False)

同じようにテストを行うとFOREIGN KEY constraint failedを吐く。
set_sqlite_pragmaはデコレーター(@event.listens_for)で接続の度に勝手に実行されるので自分で呼び出す必要はありません。

外部キーがデフォルトで有効になっていない理由

 SQLite v3.6.19まで外部キー制約が機能していなかったようです。
 https://www.sqlite.org/foreignkeys.html#fk_enable

PRAGMAについて

 SQLiteに特殊な要求を行うコマンド。今回は外部キー制約をONにするために使用している。

 テストを変更させずにエラーを吐かせる方法をご存じの方がいらっしゃいましたらご教授ください。
 @tiechel様、本当にありがとうございました。

6
2
1

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
6
2