Python AlembicでPostgres Migration管理

  • 0
    いいね
  • 0
    コメント

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

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

    ツール候補

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

    alembic参考ドキュメント

    使ってみる

    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