Python AlembicでPostgres Migration管理

Python系のPostgreSQL Databaseのmigration管理ツールを検討

public以外のスキーマも用いたい。
postgreSQLのデータ型を直接指定したい。

ツール候補

• django
• alembic ←今回はこれを利用
• sqlalchemy-migrate
• yoyo-migration

alembic参考ドキュメント

• pypi alembic
• 公式doc
• Python の Alembic を使って RDB のスキーマを管理する

使ってみる

install

$ pip install alembic

作業ディレクトリ作成

複数のDBを使うときはオプション-t multidbをつける

$ alembic init <DIRNAME>
$ alembic init <DIRNAME> -t multidb

設定ファイル編集

alembic.iniが設定ファイル。
デフォルトではディレクトリと同じ場所においておく。
他の場所におく場合は、以後の作業にオプションで-c <設定ファイルパス>と指定

データベースのconfig情報を編集

sqlalchemy.url = postgresql://<username>:<password>@<hostname>:<port>/<dbname>

alembic.iniを他の場所に移す場合は以下を編集

script_location = <DIRPATH>

新migration用スクリプト作成

$ alembic revision -m 'Create Table'

'Create Table'は作成するスクリプト名
<DIRNAME>/versions/6535ae93bef1_create_table.pyが作成される
ファイル名の6535ae93bef1はrevision ID。その都度自動で作成される。

postgreSQLのデータ型を直接指定したいので、以下をimport。

/versions/6535ae93bef1_create_table.py

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql as pg

スクリプトファイルのupgradeとdowngradeファンクションに更新内容を記載

/versions/6535ae93bef1_create_table.py

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', pg.INTEGER, primary_key=True),  # serial

        sa.Column('char_col', pg.CHAR(10),
                  nullable=False,  # default:True
                  unique=True,  # default:False
                  server_default='deftext'  # default:None
                  ),
        sa.Column('varchar_col', pg.VARCHAR(32)),
        sa.Column('text_col', pg.TEXT),

        sa.Column('bool_col', pg.BOOLEAN),

        sa.Column('smallint_col', pg.SMALLINT),
        sa.Column('int_col', pg.INTEGER, server_default='100'),
        sa.Column('double_col', pg.DOUBLE_PRECISION),
        sa.Column('numeric_col', pg.NUMERIC),
        schema='test'  # set schema name
    )

    op.create_table(
        'permissions',
        sa.Column('id', pg.INTEGER, primary_key=True),  # serial
        sa.Column('user_id', pg.INTEGER,
                  sa.ForeignKey(
                      'test.users.id', onupdate='CASCADE', ondelete='RESTRICT'
                  ),  # default ON UPDATE NO ACTION ON DELETE NO ACTION
                  ),
        sa.Column('date_col', pg.DATE, server_default=sa.text('NOW()')),  # default NOW()
        sa.Column('time_col', pg.TIME),  # time without time zone
        sa.Column('timetz_col', pg.TIME(timezone=True)),  # time with time zone
        sa.Column('timestamp_col', pg.TIMESTAMP),  # timestamp without time zone
        sa.Column('timestamptz_col', pg.TIMESTAMP(timezone=True)),  # timestamp with time zone
        sa.Column('interval_col', pg.INTERVAL),

        sa.Column('int4range_col', pg.INT4RANGE),
        sa.Column('numrange_col', pg.NUMRANGE),
        sa.Column('daterange_col', pg.DATERANGE),
        sa.Column('tsrange_col', pg.TSRANGE),
        sa.Column('tstzrange_col', pg.TSTZRANGE),

        sa.Column('json_col', pg.JSON),
        sa.Column('macaddr_col', pg.MACADDR),
        sa.Column('inet_col', pg.INET),
        schema='test'
    )


def downgrade():
    op.drop_table('account', schema='test')

testスキーマは残念ながら自動で作成されないなので、事前に作成しておく。

CREATE SCHEMA test;

upgrade

最新に更新

$ alembic upgrade head

一つ戻る

$ alembic downgrade -1

初期状態に戻る

$ alembic downgrade base

状態確認

現在の状態

$ alembic current

ヒストリ

$ alembic history