Zenn
🐋

alembicのマイグレーションファイルの書き方

2023/09/26に公開
Python
alembic
tech

Pythonのマイグレーションツールalembicで、テーブル作成や列追加等をどう記載すればよいかまとめます。
alembicのインストールや設定方法については触れません。

テーブル作成

from datetime import datetime

from alembic import op
from sqlalchemy import DATETIME, VARCHAR, Column, Integer


def upgrade() -> None:
    op.create_table(
        "users",
        Column("id", Integer, primary_key=True),
        Column("name", VARCHAR(255), nullable=False),
        Column("email", VARCHAR(255), nullable=False),
        Column("deleted_at", DATETIME, default=None),
        Column("created_at", DATETIME, default=datetime.now, nullable=False),
        Column(
            "updated_at",
            DATETIME,
            default=datetime.now,
            nullable=False,
            onupdate=datetime.now,
        ),
    )

テーブル削除

from alembic import op


def upgrade() -> None:
    op.drop_table("users")

列追加

from alembic import op
from sqlalchemy import VARCHAR, Column
from sqlalchemy.dialects.mysql import TINYINT


def upgrade() -> None:
    op.add_column("users", Column("deleted", TINYINT))

列削除

from alembic import op


def upgrade() -> None:
    op.drop_column("users", "email")

列定義変更

import sqlalchemy as sa
from alembic import op


def upgrade() -> None:
    op.alter_column(
        "users",
        "name",
        existing_type=sa.VARCHAR(255),
        type_=sa.VARCHAR(20),
        nullable=False,
    )

任意のSQL実行

sa.text()を使うことで、文字列として記載したSQLの実行も可能です。変数を使用する場合は:変数名として、.bindparams()で値を設定する。
いくつか例を記載します。

insert

import sqlalchemy as sa
from alembic import op


def upgrade() -> None:
    op.execute(
        sa.text(
            """
            INSERT INTO users (id, name, email, created_at, updated_at)
            VALUES (:id, :name, :email, :created_at, :updated_at)
        """
        ).bindparams(
            sa.bindparam("id", 1),
            sa.bindparam("name", "John Smith"),
            sa.bindparam("email", "john@example.com"),
            sa.bindparam("created_at", "2000-01-01"),
            sa.bindparam("updated_at", "2000-01-01"),
        )
    )

カラム追加(位置指定)

def upgrade() -> None:
    op.execute(
        sa.text(
            """
            ALTER TABLE users ADD deleted TINYINT(1) NOT NULL DEFAULT 0
	    AFTER email
        """
        )

Discussion