[SQLAlchemy] CASCADEを使用して親テーブルが削除された時、子テーブルも削除されるようにする（テスト実装あり）

# この記事を読んでできること
参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除されるようにする実装ができる
# この記事の概要
1. CASCADE実装
1. 正常な挙動がされるか（テスト実装）
# 記述しないこと
- 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のリレーションにおけるメソッドやパラメータについてのメモ](https://poyo.hatenablog.jp/entry/2017/01/08/212227#cascade)

これで参照整合性制約があっても、親テーブルが削除された時、小テーブルも削除される
# ほな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
```
## この実装を通しての所感
テストが非常に苦労しました。
データベースを無差別に創造して破壊しまくってました。
こんなに長いテストなのにたったの１ケースです。。。
いい勉強になりました！
テストはかなり冗長な気もするので、何かお気づきの点等あればお気軽にコメントください！

# 参考
- [外部キーと参照整合性を理解しよう](https://itmanabi.com/foreign-key-referential-integrity/)
- [SQLAlchemy Cascades](https://docs-sqlalchemy.readthedocs.io/ko/latest/orm/cascades.html)
- [SQLAlchemyのリレーションにおけるメソッドやパラメータについてのメモ](https://poyo.hatenablog.jp/entry/2017/01/08/212227#cascade)
- [MySQLの外部キー制約RESTRICT,CASCADE,SET NULL,NO ACTIONの違いは？](https://qiita.com/suin/items/21fe6c5a78c1505b19cb#on-delete-cascade)