使用Alembic更改Enum字段

Question

当使用早于9.1的PostgreSQL版本(为枚举添加ALTER TYPE)时,如何在alembic迁移中向Enum字段添加元素？这个 SO问题解释了直接过程,但我不太确定如何使用alembic进行翻译.

这就是我所拥有的:

new_type = sa.Enum('nonexistent_executable', 'output_limit_exceeded',
                   'signal', 'success', 'timed_out', name='status')
old_type = sa.Enum('nonexistent_executable', 'signal', 'success', 'timed_out',
                   name='status')
tcr = sa.sql.table('testcaseresult',
                   sa.Column('status', new_type, nullable=False))


def upgrade():
    op.alter_column('testcaseresult', u'status', type_=new_type,
                    existing_type=old_type)


def downgrade():
    op.execute(tcr.update().where(tcr.c.status==u'output_limit_exceeded')
               .values(status='timed_out'))
    op.alter_column('testcaseresult', u'status', type_=old_type,
                    existing_type=new_type)

不幸的是,上面只ALTER TABLE testcaseresult ALTER COLUMN status TYPE status在升级时产生,基本上什么也没做.

Answer 1

我决定尝试尽可能直接遵循postgres方法并提出以下迁移.

from alembic import op
import sqlalchemy as sa

old_options = ('nonexistent_executable', 'signal', 'success', 'timed_out')
new_options = sorted(old_options + ('output_limit_exceeded',))

old_type = sa.Enum(*old_options, name='status')
new_type = sa.Enum(*new_options, name='status')
tmp_type = sa.Enum(*new_options, name='_status')

tcr = sa.sql.table('testcaseresult',
                   sa.Column('status', new_type, nullable=False))


def upgrade():
    # Create a tempoary "_status" type, convert and drop the "old" type
    tmp_type.create(op.get_bind(), checkfirst=False)
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status TYPE _status'
               ' USING status::text::_status')
    old_type.drop(op.get_bind(), checkfirst=False)
    # Create and convert to the "new" status type
    new_type.create(op.get_bind(), checkfirst=False)
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status TYPE status'
               ' USING status::text::status')
    tmp_type.drop(op.get_bind(), checkfirst=False)


def downgrade():
    # Convert 'output_limit_exceeded' status into 'timed_out'
    op.execute(tcr.update().where(tcr.c.status==u'output_limit_exceeded')
               .values(status='timed_out'))
    # Create a tempoary "_status" type, convert and drop the "new" type
    tmp_type.create(op.get_bind(), checkfirst=False)
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status TYPE _status'
               ' USING status::text::_status')
    new_type.drop(op.get_bind(), checkfirst=False)
    # Create and convert to the "old" status type
    old_type.create(op.get_bind(), checkfirst=False)
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status TYPE status'
               ' USING status::text::status')
    tmp_type.drop(op.get_bind(), checkfirst=False)

似乎alembic USING对其alter_table方法中的陈述没有直接支持.

Answer 2

我使用了一个更简单的方法,步骤少于接受的答案,我基于此.在这个例子中,我将假装有问题的枚举被称为'status_enum',因为在接受的答案中,对列和枚举使用'status'使我感到困惑.

from alembic import op 
import sqlalchemy as sa

name = 'status_enum'
tmp_name = 'tmp_' + name

old_options = ('nonexistent_executable', 'signal', 'success', 'timed_out')
new_options = sorted(old_options + ('output_limit_exceeded',))

new_type = sa.Enum(*new_options, name=name)
old_type = sa.Enum(*old_options, name=name)

tcr = sa.sql.table('testcaseresult',
                   sa.Column('status', new_type, nullable=False))

def upgrade():
    op.execute('ALTER TYPE ' + name + ' RENAME TO ' + tmp_name)

    new_type.create(op.get_bind())
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status ' +
               'TYPE ' + name + ' USING status::text::' + name)
    op.execute('DROP TYPE ' + tmp_name)


def downgrade():
    # Convert 'output_limit_exceeded' status into 'timed_out'                                                                                                                      
    op.execute(tcr.update().where(tcr.c.status=='output_limit_exceeded')
               .values(status='timed_out'))

    op.execute('ALTER TYPE ' + name + ' RENAME TO ' + tmp_name)

    old_type.create(op.get_bind())
    op.execute('ALTER TABLE testcaseresult ALTER COLUMN status ' +
               'TYPE ' + name + ' USING status::text::' + name)
    op.execute('DROP TYPE ' + tmp_name)

Answer 3

从Postgres 9.1开始,可以使用ALTER TYPE语句向枚举添加新值.由于无法在事务中完成,因此这很复杂.然而,这可以通过提交alembic的交易来解决.

我实际上在使用旧的,更详细的解决方案时遇到了问题,因为Postgres无法自动转换列的默认值.

Answer 4

首先将列类型更改为 VARCHAR()。

然后删除您的类型并使用新字段创建新类型。

最后将列类型更改为新创建的类型。

def upgrade():
    op.execute(
        '''
        ALTER TABLE your_table ALTER COLUMN your_enum_column TYPE VARCHAR(255);

        DROP TYPE IF EXISTS your_enum_type;

        CREATE TYPE your_enum_type AS ENUM 
            ('value1', 'value2', 'value3', 'value4');

        ALTER TABLE your_table ALTER COLUMN your_enum_column TYPE your_enum_type 
            USING (your_enum_column::your_enum_type);
        '''
    )


def downgrade():
    op.execute(
        '''
        ALTER TABLE your_table ALTER COLUMN your_enum_column TYPE VARCHAR(255);

        DROP TYPE IF EXISTS your_enum_type;

        CREATE TYPE your_enum_type AS ENUM 
            ('value1', 'value2', 'value3');

        ALTER TABLE your_table ALTER COLUMN your_enum_column TYPE your_enum_type 
            USING (your_enum_column::your_enum_type);
        '''
    )

Answer 5

运行没有问题：

from alembic import op

def upgrade():
    op.execute("COMMIT")
    op.execute("ALTER TYPE enum_type ADD VALUE 'new_value'")

def downgrade():
    ...

参考

Answer 6

我在尝试将列类型迁移到另一个时遇到了同样的问题。我使用以下要求：

Alembic==0.9.4
SQLAlchemy==1.1.12 

您可以将参数提供postgresql_using为alembic.op.alter_column。

from alembic import op
import sqlalchemy as types

op.alter_column(
    table_name='my_table',
    column_name='my_column',
    type_=types.NewType,
    # allows to use postgresql USING
    postgresql_using="my_column::PostgesEquivalentOfNewType",
)

希望对您有所帮助。