主页 / coding / 问题 / 79348917
Accepted
jaspionU2
Asked: 2025-01-12 05:18:40 +0800 CST

如何使用 Alembic 将列从 ENUM 更改为 INTEGER（外键）？

  • 772

我正在尝试使用 Alembic 将 PostgreSQL 中的列类型从 ENUM 更改为 INTEGER（这也是外键）。但是，当我尝试升级迁移时，遇到以下错误：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.CannotCoerce) 无法将类型 serie_enum 转换为整数 LINE 1: ...R COLUMN serie_aluno TYPE INTEGER USING serie_aluno::integer
[SQL: ALTER TABLE aluno ALTER COLUMN serie_aluno TYPE INTEGER USING serie_aluno::integer]

以下是 Alembic 生成的迁移文件：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('aluno', 'serie_aluno',
               existing_type=postgresql.ENUM('6', '7', '8', '9', name='serie_enum'),
               type_=sa.Integer(),
               existing_nullable=False,
               postgresql_using="NULL")
    op.create_foreign_key(None, 'aluno', 'serie', ['serie_aluno'], ['id_serie'])
    # ### end Alembic commands ###

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'aluno', type_='foreignkey')
    op.alter_column('aluno', 'serie_aluno',
               existing_type=sa.Integer(),
               type_=postgresql.ENUM('6', '7', '8', '9', name='serie_enum'),
               existing_nullable=False)
    # ### end Alembic commands ###

如何才能成功地将列从 ENUM 转换为 INTEGER，同时确保正确创建外键约束？

sql

3 个回答

  1. 对于任何感兴趣的人来说，解决方案实际上非常简单，基本上，我按照我的朋友 Erwin 的建议添加了 sql 命令。

    def upgrade() -> None:

  op.execute("ALTER TABLE aluno ALTER COLUMN serie_aluno TYPE integer USING serie_aluno::text::int, ADD CONSTRAINT serie_aluno_serie_fkey FOREIGN KEY (serie_aluno) REFERENCES serie(id_serie);")

def downgrade() -> None:

  op.drop_constraint('serie_aluno_serie_fkey', 'aluno', type_='foreignkey')
  op.alter_column(
        'aluno',
        'serie_aluno',
        type_=sa.Enum('A', 'B', name='turma_enum'),
        postgresql_using="serie_aluno::int::text"
        )
    • 2
  2. Best Answer

    没有定义从自定义枚举类型serie_aluno到的强制类型转换integer。（除非您明确创建这样的强制类型转换。）

    但一切都可以转换为text。如果所有枚举值都由有效的整数文字表示，则可以使用text作为垫脚石：

    ALTER TABLE aluno ALTER COLUMN serie_aluno TYPE int USING serie_aluno::text::int;

    然后添加指向某一列的约束：FOREIGN KEYinteger

    ALTER TABLE aluno ADD CONSTRAINT aluno_serie_fkey FOREIGN KEY (serie_aluno) REFERENCES serie(id_serie);

    或者在单个ALTER TABLE命令中执行此操作：

    ALTER TABLE aluno
  ALTER COLUMN serie_aluno TYPE integer USING serie_aluno::text::int
, ADD CONSTRAINT serie_aluno_serie_fkey FOREIGN KEY (serie_aluno) REFERENCES serie(id_serie)
;

    小提琴

    • 1

  3. 这有点混乱，但我必须先通过一次更改使该列可空，然后才能进行后续更改以更改类型并将所有值设置为。如果您知道映射并使用您说不需要但我认为很酷的语句，NULL您实际上可以将枚举迁移到硬编码整数。CASE

    • 创建虚拟数据
    • 将列更改为空
    • 将列的类型更改为 int
    • 创建外键
    • 设置单个 fk 值
    • 检查关系是否正常
    import os

from sqlalchemy import (
    create_engine,
    Column,
    Integer,
    String,
    ForeignKey,
    DateTime,
)
import alembic
from sqlalchemy.orm import Mapped, aliased
from sqlalchemy.orm import mapped_column
from sqlalchemy.sql import (
    text,
    select,
    func,
    and_,
    or_,
    update,
    insert,
    delete,
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Session,
    relationship,
    selectinload,
    joinedload,
)
from typing import List
from sqlalchemy import event
from sqlalchemy.schema import CreateSchema
from sqlalchemy.dialects import postgresql
from alembic.migration import MigrationContext
from alembic.operations import Operations


def get_engine(env):
    return create_engine(f"postgresql://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}", echo=True)



def get_meta(with_enum=True):

    class Base(DeclarativeBase):
        pass

    # I am sure there is a smarter way to do this but for now
    # I just redefine the entire metadata after the migration.
    if with_enum:
        class Serie(Base):
            __tablename__ = 'series'
            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()
            serie_aluno: Mapped[postgresql.ENUM] = mapped_column(postgresql.ENUM('6', '7', '8', '9', name='serie_enum'), nullable=False)
    else:
        class Serie(Base):
            __tablename__ = 'series'
            id: Mapped[int] = mapped_column(primary_key=True)
            name: Mapped[str] = mapped_column()
            serie_aluno: Mapped[int] = mapped_column(ForeignKey('alunos.id'))
            aluno: Mapped["Aluno"] = relationship("Aluno")

    class Aluno(Base):
        __tablename__ = 'alunos'
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str]

    return Base, Serie, Aluno


def main(env):
    engine = get_engine(env)

    Base, Serie, Aluno = get_meta(with_enum=True)
    with engine.connect() as conn, conn.begin():

        Base.metadata.create_all(conn)

    with Session(engine) as db:
        db.add(Serie(name='serie1', serie_aluno='6'))
        db.add(Serie(name='serie2', serie_aluno='9'))
        db.add(Aluno(name='aluno1'))
        db.commit()

    with engine.connect() as conn:
        with conn.begin() as trans:
            context = MigrationContext.configure(conn)
            op = Operations(context)
            #whens = ' '.join(f"WHEN serie_aluno='{v}' THEN {v}" for v in ('6', '7', '8', '9'))
            #postgresql_using=f"CASE {whens} END"
            # Change from nullable = False to nullable = True
            op.alter_column(Serie.__tablename__, "serie_aluno", nullable=True, existing_nullable=False)
            # Alter column and set values to NULL
            op.alter_column(Serie.__tablename__, "serie_aluno", existing_type=postgresql.ENUM, type_=Integer, postgresql_using="NULL")
            op.create_foreign_key(None, 'series', 'alunos', ['serie_aluno'], ['id'])
            trans.commit()

    Base, Serie, Aluno = get_meta(with_enum=False)

    with Session(engine) as db:
        db.execute(text("UPDATE series SET serie_aluno = 1 WHERE name = 'serie1'"))
        db.flush()
        assert db.scalar(select(func.count(Serie.id)).where(Serie.serie_aluno != None)) == 1



if __name__ == '__main__':
    main(os.environ)
    • 0

相关问题