この記事を読んでできること
参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除されるようにする実装ができる
この記事の概要
- CASCADE実装
- 正常な挙動がされるか（テスト実装）
記述しないこと
- SQLAlchemyとは
- 基本的なモデル定義
- 基本的なpytestを使用したテスト方法
- conftest.pyについて
仕様技術
- fastapi==0.78.0
- mysqlclient==2.1.0
- pydantic==1.9.1
- python-dotenv==0.20.0
- requests==2.28.0
- SQLAlchemy==1.4.36
- SQLAlchemy-Utils==0.38.2
- uvicorn==0.17.6
- pytest==7.1.2
そもそもCASCADEとは？
- 外部キー制約（参照整合性）におけるオプションの一つで、依存関係を持った親と子のレコード同士の整合性を保つために、親への操作を子のレコードにも一貫して操作を連動させるような仕組みのこと
- 具体的には、親テーブルのレコードを削除または更新したときに、子テーブルの関連するレコードを自動的に削除または更新してくれる機能のこと
実装してみよう
まずはコード例（親テーブル）
models/parent.py
from sqlalchemy import BigInteger, Column, String
# Baseクラス作成用にインポート
from sqlalchemy.ext.declarative import declarative_base
# mysqlのutf8は3バイト文字に制限されている為、4バイト文字も処理するutf8mb4をデフォルトで使用する。
class Base(object):
__table_args__ = {"mysql_default_charset": "utf8mb4"}
# Baseクラスを作成
Base = declarative_base(cls=Base)
class Parent(Base):
__tablename__ = "parent"
id = Column(BigInteger, primary_key=True, nullable=False)
last_name = Column(String(20), nullable=False)
first_name = Column(String(20), nullable=False)
コード例（子テーブル）
models/child.py
# Baseクラス作成用にインポート
from sqlalchemy import BigInteger, Column, ForeignKey, String
from sqlalchemy.sql import func
# Baseクラス作成用にインポート
from sqlalchemy.ext.declarative import declarative_base
# mysqlのutf8は3バイト文字に制限されている為、4バイト文字も処理するutf8mb4をデフォルトで使用する。
class Base(object):
__table_args__ = {"mysql_default_charset": "utf8mb4"}
# Baseクラスを作成
Base = declarative_base(cls=Base)
from models.parent import Parent
class Child(Base):
__tablename__ = "child"
id = Column(BigInteger, primary_key=True, nullable=False)
parent_id = Column(
BigInteger, ForeignKey(Parent.id, ondelete="CASCADE"), nullable=False
)
company_name = Column(String(20), nullable=False)
- Childモデルのparent_idカラムに
ForeignKey(Parent.id, ondelete="CASCADE")の引数を持たせることでCASCADEの設定は完了 - マイグレーション時に設定に反映されていることがわかる
↓childテーブルのマイグレーションログ
FOREIGN KEY(parent_id) REFERENCES parent (id) ON DELETE CASCADE
SQLAlchemyではdelete以外にも多岐にわたって処理を記述できる↓
これで参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除される
ほなPytestでテスト書いてみようか
conftest.py
import pytest
import requests
import codecs
import os
from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy_utils import create_database, database_exists
from fastapi.testclient import TestClient
from dotenv import load_dotenv
from models.parent import Parent
from models.child import Child
from main import app
from routers.router import get_db
load_dotenv()
@pytest.fixture
def application_fixture():
json = {
"last_name": "sota",
"first_name": "heavymetal21",
"company_name": "Rabbit Company",
}
return json
@pytest.fixture()
def test_db_creation_and_deletion():
# そのままでは「utf8mb4」を認識しないので、codecsモジュールを使用して、utf8mb4でファイルを開く
codecs.register(lambda name: codecs.lookup("utf8") if name == "utf8mb4" else None)
# 接続したいDBへの接続情報
connection_url = URL.create(
drivername="mysql",
username=os.environ["MYSQL_USER"],
password=os.environ["MYSQL_PASSWORD"],
host=os.environ["MYSQL_HOST"],
database=os.environ["MYSQL_TEST_DATABASE"],
port=os.environ["MYSQL_PORT"],
query={"charset": "utf8mb4"},
)
# DBとの接続
ENGINE = create_engine(
connection_url,
# 文字コードを指定
encoding="utf8mb4",
# 自動生成されたSQLを吐き出すようにする
echo=True,
hide_parameters=True,
)
# session変数にsessionmakerインスタンスを格納
test_session = scoped_session(
# ORマッパーの設定。自動コミットと自動反映はオフにする
sessionmaker(autocommit=False, autoflush=False, bind=ENGINE)
)
# DBが存在しなければ
if not database_exists(ENGINE.url):
# DBを新規作成する
create_database(ENGINE.url)
# 定義されているテーブルを作成
Parent.metadata.create_all(bind=ENGINE)
Child.metadata.create_all(bind=ENGINE)
# DB接続用のセッションクラス、インスタンスが作成されると接続する
Parent.query = test_session.query_property()
Child.query = test_session.query_property()
# テスト用のsessionの依存関係を定義する
def override_get_db():
try:
db = test_session()
yield db
finally:
db.close()
# テスト用のoverride_get_dbでget_dbをオーバーライドする(重要だよ)
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
yield {"client": client, "test_session": test_session}
# テスト終了時にテスト用DBを削除する
test_session.execute("drop database test_pc_app_form;")
# 元のget_dbに戻す
app.dependency_overrides[get_db] = get_db
テスト時のみ、テスト用のget_db関数を作成し、オーバーライドすることでテスト用のデータベースを利用できる
こちらの公式ドキュメントが非常に分かりやすい
https://fastapi.tiangolo.com/ja/advanced/testing-database/
test_db_creation_and_deletion.py
# CASCADEによってParentテーブルが削除されたとき、他の全てのテーブルが削除されるかどうか
def test_delete_table_on_cascade(application_fixture, test_db_creation_and_deletion):
client = test_db_creation_and_deletion["client"]
session = test_db_creation_and_deletion["test_session"]
# テーブルにデータを格納する
# /api/applicationは申し込みをするAPIです
client.post("/api/application", json=application_fixture)
# selectのクエリ
count_parent_query = "SELECT COUNT(*) FROM parent;"
count_child_query = "SELECT COUNT(*) FROM child;"
# selectのクエリを実行
parent = session.execute(count_parent_query)
child = session.execute(count_child_query)
# 各テーブルの行数を数える
count_parent = parent.fetchone()[0]
count_child = child.fetchone()[0]
# 期待する行数
expected = 1
assert count_parent == expected
assert count_child == expected
# parentテーブルのデータを削除するクエリを実行
delete_parent_query = "DELETE FROM parent;"
session.execute(delete_parent_query)
# selectのクエリを実行
parent_after_delete = session.execute(count_parent_query)
child_after_delete = session.execute(count_child_query)
# 各テーブルの行数を数える
count_parent_after_delete = parent_after_delete.fetchone()[0]
count_child_after_delete = child_after_delete.fetchone()[0]
expected_after_delete = 0
# 各テーブルの行数がCASCADE機能によって０になっている
assert count_parent_after_delete == expected_after_delete
assert count_child_after_delete == expected_after_delete
この実装を通しての所感
テストが非常に苦労しました。
データベースを無差別に創造して破壊しまくってました。
こんなに長いテストなのにたったの１ケースです。。。
いい勉強になりました！
テストはかなり冗長な気もするので、何かお気づきの点等あればお気軽にコメントください！