Missing batch_op.f()

[quentin-roche]

Describe the bug

When alembic generates the code for migrations, it uses op.f("constraint_name") to encapsulate contraint names. The goal is to make sure that constraint names get truncated the same way as SQLAlchemy does (truncate_and_render_constraint_name). SQLAlchemy truncate names because MySQL does not accept contraint names that are more than 64 character long.

I don't know why (did not have the time to search for too long) but sometimes Alembic will not add batch_op.f() in batch operation. It seems to me that it does not add during:

• Drop of constraint during upgrade
• Creation of constraint during downgrade

This behaviour of not adding the batch_op.f() seem to affect:

• DropConstraintOp
• DropIndexOp
• AddConstraintOp
• CreateForeignKeyOp
• Constraint
• maybe other things as weel

Expected behavior

Adding op.f("constraint_name") to truncate all contraint names.

To Reproduce

I can't share the full code but this is my env file:

def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        render_as_batch=True,
    )

    with context.begin_transaction():
        logging.warning("Offline migration")
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            render_as_batch=True,
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Error

I don't remember if it was an error or the contraint being recreating for each migration. (It was few weeks ago)

Versions

• OS: MacOS
• Python: 3.11
• Alembic: alembic --version > alembic 1.13.1
• SQLAlchemy: sqlalchemy == 2.0.27
• Database: From memory, I was generating the migration using SQLite. But I may also have tried with MySQL as well
• DBAPI: both mysql and sqlite

Dirty fix

I had many old migration that were affected and did not want to modify manually thousands of lines of codes. The fix is more for people that already have migrations with missing batch_op.f() and don't want to modify them.

def fix_names_not_encapsulated(
    func, arg_pos: int | None = None, kwarg_key: str | None = None
):
    def sanitize_name(constraint_name: str | None) -> str | None | sqlalchemy_conv:
        if not (
            constraint_name is None
            or constraint_name == "alembic_version_pkc"
            or isinstance(constraint_name, sqlalchemy_conv)
        ):
            constraint_name = op.f(constraint_name)

            for frame_info in inspect.stack()[1:]:
                if os.path.normpath(os.path.dirname(__file__)) in frame_info.filename:
                    message = (
                        f'Missing encapsulation for "{constraint_name}" '
                        f'(File "{frame_info.filename}", '
                        f"line {frame_info.lineno}, "
                        f"in {frame_info.function})"
                    )
                    break
            else:
                message = f"Missing encapsulation for {constraint_name}"
            logging.info(message)
        return constraint_name

    @wraps(func)
    def wrapper(*args, **kwargs):
        if arg_pos is not None:
            args = (*args[:arg_pos], sanitize_name(args[arg_pos]), *args[arg_pos + 1 :])
        elif kwarg_key is not None:
            kwargs[kwarg_key] = sanitize_name(kwargs[kwarg_key])
        else:
            raise ValueError("You must specify either arg_pos or kwarg_key")
        func(*args, **kwargs)

    return wrapper


DropConstraintOp.__init__ = fix_names_not_encapsulated(DropConstraintOp.__init__, arg_pos=1)
DropIndexOp.__init__ = fix_names_not_encapsulated(DropIndexOp.__init__, arg_pos=1)
CreateIndexOp.__init__ = fix_names_not_encapsulated(CreateIndexOp.__init__, arg_pos=1)
AddConstraintOp.__init__ = fix_names_not_encapsulated(AddConstraintOp.__init__, arg_pos=1)
CreateForeignKeyOp.__init__ = fix_names_not_encapsulated(CreateForeignKeyOp.__init__, arg_pos=1)
Constraint.__init__ = fix_names_not_encapsulated(Constraint.__init__, kwarg_key="name")

If you need some more info, feel free. This is not really an issue for me anymore since I got a fix but it might help someone understand what is going wrong with its own setup

Have a nice day!

[zzzeek]

you dont need op.f() if your constraints have names already or if you arent using naming conventions, can I see an example of a "bad" migration as well as your naming convention setup? thanks

[zzzeek]

The goal is to make sure that constraint names get truncated the same way as SQLAlchemy does (truncate_and_render_constraint_name). SQLAlchemy truncate names because MySQL does not accept contraint names that are more than 64 character long.

that's not the purpose of op.f(). op.f() is strictly to override constraint naming conventions from taking place. if your names are hardcoded more than 64 chars that's a bug on your end and you need to fix that

[quentin-roche]

I'm using naming convention. I did not create the names. This is my SQLAlchemy setup:

from sqlalchemy.orm.decl_api import declarative_base
from sqlalchemy.schema import MetaData

convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}
Base = declarative_base(metadata=MetaData(naming_convention=convention))

metadata: MetaData = Base.metadata

When using metadata.create_all the names in the database are correctly truncated. But this is part of one of the downgrade generated by Alembic:

def upgrade() -> None:
   [...]
    with op.batch_alter_table("single_source_of_truth", schema=None) as batch_op:
        batch_op.add_column(
            sa.Column("asset_models_version_id", sa.Integer(), nullable=True)
        )
        batch_op.add_column(
            sa.Column("asset_financials_version_id", sa.Integer(), nullable=True)
        )
        batch_op.add_column(
            sa.Column("financial_context_version_id", sa.Integer(), nullable=True)
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_solar_pv_version_id_file_version",
            type_="foreignkey",
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_biogas_plant_version_id_file_version",
            type_="foreignkey",
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_air_separation_unit_version_id_file_version",
            type_="foreignkey",
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_saf_fischer_tropsch_version_id_file_version",
            type_="foreignkey",
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_desalination_plant_version_id_file_version",
            type_="foreignkey",
        )
        batch_op.drop_constraint(
            "fk_single_source_of_truth_hydrogen_longterm_storage_version_id_file_version",

print(len('fk_single_source_of_truth_hydrogen_longterm_storage_version_id_file_version'))
75

It might have something to do with the fact that I may have use SQLite during the migration generation.

[zzzeek]

the f() operation can always be from op.f() and it's not needed for those FK names. op.f() is only needed when the naming convention uses the "%(constraint_name)s" token, to indicate this name is already translated.

names like ""fk_single_source_of_truth_solar_pv_version_id_file_version" that are DROPs here came directly from your database, which in this case would be the SQLite database. So if those names are then too long for MySQL, this is not a directly supported use case for alembic, alembic assumes autogen migrations would be created against the specific type of database that is used for production operations. so you would need to manually specify these constraint names in your code.

now after all of that, it just so happens that the op.f() construct is a subclass of SQLAlchemy's truncated label construct, which is private. it's in fact using conv. none of the docs here indicate that this is also useful to apply truncation rules to an identifier, so at the moment that's a happenstance side effect of what's going on.

So what this implies is, there's a documentation bug and test coverage bug in SQLAlchemy which is that the conv construct happens to also apply name truncation rules to strings, we dont document this, and most importantly we dont test this, meaning, I could change it someday, no tests would break, and all the code out there relying on this effect would break.

so here is the bug: sqlalchemy/sqlalchemy#11284

after all of that, op.f() has apparent other behaviors that are useful. i think this falls under #264. that's an open issue, and a lot of problems were produced when we did it.

TLDR, apply op.f() and/or sqlalchemy.conv to names you'd like truncated from SQLite's char length to MySQLs.

[quentin-roche]

Ok I see. Thank you for your help !