【備忘録】PythonでMySQLのデータベースをマイグレーションする方法

備考

Python:3.9
MySQL:8.0.31
動作環境:MacOS
開発環境:PyCharm

事前準備

データベース作成済み

1. PyCharmにて新規プロジェクトを作成する

2. プロジェクトを以下の構成になるようにファイルを作成する

フォルダ構成.

.
├── main.py
├── env.py
├── .env
├── requirements.txt
└── database
    ├── database.py
    └── models.py

3. requirements.txtファイルを以下の内容にする

requirements.txt

alembic
pytz
python-dotenv

4. ターミナルにてコマンドを入力してモジュールをインストールする

ターミナル.

pip install -r requirements.txt

5. .envファイルを以下の内容にする

.env

# DB接続情報
DATABASE = 'mysql'
DB_USER = 'root'
DB_PASSWORD = 'MySQLのパスワード'
DB_HOST = 'localhost'
DB_PORT = '3306'
DB_NAME = '対象のDB名'

6. env.pyファイルを以下の内容にする

env.py

import os
from dotenv import load_dotenv
from os.path import join, dirname

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)


class Env:
    # DB接続情報
    DATABASE = os.environ.get("DATABASE")
    DB_USER = os.environ.get("DB_USER")
    DB_PASSWORD = os.environ.get("DB_PASSWORD")
    DB_HOST = os.environ.get("DB_HOST")
    DB_PORT = os.environ.get("DB_PORT")
    DB_NAME = os.environ.get("DB_NAME")

7. database.pyファイルを以下の内容にする

database.py

from env import Env
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

# データベースに接続するための情報
database_url = '{}://{}:{}@{}:{}/{}?charset=utf8'.format(Env.DATABASE, Env.DB_USER, Env.DB_PASSWORD, Env.DB_HOST, Env.DB_PORT, Env.DB_NAME)
engine = create_engine(database_url, connect_args={"connect_timeout": 15}, echo=False, pool_recycle=10, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# データベースに接続するための処理
def db_session():
    return scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

8. models.pyファイルを以下の内容にする

models.py

from database.database import Base
from datetime import datetime
from sqlalchemy import Column, VARCHAR, INT, TEXT, DATETIME
import pytz

jst = pytz.timezone('Asia/Tokyo')


# ユーザーのモデルを用意
class User(Base):
    __tablename__ = 'users'
    __table_args__ = {"mysql_collate": "utf8_general_ci"}
    id = Column(INT, primary_key=True, autoincrement=True, unique=True, nullable=False, index=True)
    name = Column(VARCHAR(255), primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    email = Column(VARCHAR(255), primary_key=False, autoincrement=False,  unique=True, nullable=False, index=True)
    password = Column(TEXT, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    create_time = Column(DATETIME, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False, default=lambda: datetime.now(jst)) 

9. ターミナルにてコマンドを入力してマイグレーションフォルダを生成する

ターミナル.

alembic init migrations

10. プロジェクト直下にmigrations / alembic.iniが生成される

フォルダ構成.

.
├── main.py
├── env.py
├── .env
├── requirements.txt
├── alembic.ini
├── database
│   ├── database.py
│   └── models.py
└── migrations
    ├── README
    ├── env.py
    ├── script.py.mako
    └── versions

11. migrations/env.pyファイルを以下の内容にする

env.py

from sqlalchemy import pool
from sqlalchemy import engine_from_config
from alembic import context
from database import database, models
from logging.config import fileConfig

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# target_metadata = database.Base.metadata # どちらかを使用する
target_metadata = models.Base.metadata     # どちらかを使用する

config.set_main_option("sqlalchemy.url", str(database.database_url))


def run_migrations_offline() -> None:

    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:

    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online() 

12. main.pyファイルを以下の内容にする

main.py

import subprocess

if __name__ == "__main__":
    print("データベースのマイグレーションを開始します")

    print("新規マイグレーションスクリプトを生成します")
    cmd = "alembic revision --autogenerate"
    subprocess.call(cmd.split(" "))
    print("マイグレーションスクリプトの生成が終了しました")

    print("データベースのスキーマを最新の状態に更新します")
    cmd = "alembic upgrade head"
    subprocess.call(cmd.split(" "))
    print("データベースのスキーマを最新の状態に更新しました")

    print("データベースのマイグレーションが完了しました")

13. ターミナルにてコマンドを入力してマイグレーションを実行する

ターミナル.

python main.py

14. migrations/versions直下にマイグレーションスクリプトが生成される

フォルダ構成.

.
├── main.py
├── migration_database.py
├── env.py
├── .env
├── requirements.txt
├── alembic.ini
├── database
│   ├── database.py
│   └── models.py
└── migrations
    ├── README
    ├── env.py
    ├── script.py.mako
    └── versions
        └── マイグレーションスクリプト