この記事を読んでできること

参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除されるようにする実装ができる

この記事の概要

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以外にも多岐にわたって処理を記述できる↓

SQLAlchemyのリレーションにおけるメソッドやパラメータについてのメモ

これで参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除される

ほな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

この実装を通しての所感

テストが非常に苦労しました。
データベースを無差別に創造して破壊しまくってました。
こんなに長いテストなのにたったの１ケースです。。。
いい勉強になりました！
テストはかなり冗長な気もするので、何かお気づきの点等あればお気軽にコメントください！

参考