Edited at

Python AlembicでPostgres Migration管理

More than 1 year has passed since last update.

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