pythonpostgresqlsqlalchemyalembic

How to use `alembic upgrade head` while requesting DB commit in between each file?


Using Alembic updating an ENUM TYPE I found myself blocked because the datamigration didn't want to use the enum new values without a commit in between files.

I tried to force the commit in the migration with no luck. And I finally ran alembic twice in order to fix it.

(alembic upgrade +1 && alembic upgrade head) || exit 0

Which is not to my liking.

ChatGPT suggest to use a flag that doesn't exist but would be exactly what I'm looking for.

alembic upgrade --commit head

Do you know how I can run alembic upgrade head while ensuring that commits are made between each migration file?

Refs New enum values must be committed before they can be used


Solution

  • As described by Adrian, the way to fix this is to encapsulate the ALTER TYPE within an autocommit block:

        with op.get_context().autocommit_block():
            op.execute("ALTER TYPE rolename ADD VALUE IF NOT EXISTS 'OWNER'")
    

    As a bonus, here is a little utility that you can use for ENUM updates:

    from enum import Enum
    
    from alembic import op
    
    
    def generate_enum_add_value_sql(enum_cls: type[Enum], enum_type_name: str, new_values: set[str]) -> list[str]:
        """Generate ALTER TYPE statements to add missing enum values."""
        sql_statements = []
        for member in enum_cls:
            # If new_values is specified only create new values
            if not new_values or member.value in new_values:
                sql_statements.append(
                    f"ALTER TYPE {enum_type_name} ADD VALUE IF NOT EXISTS '{member.value.upper()}'"
                )
        return sql_statements
    
    
    def execute_enum_migration(
        enum_cls: type[Enum], enum_type_name: str, new_values: set[str] | None = None
    ) -> None:
        """Execute ALTER TYPE statements for all values in enum_cls not already in the DB."""
        if new_values is None:
            new_values = set()
    
        with op.get_context().autocommit_block():
            for sql in generate_enum_add_value_sql(enum_cls, enum_type_name, new_values):
                op.execute(sql)