Alembic Drops and Recreates Existing Foreign Key with Same Name When Adding Another Column

[lclbm]

from typing import TYPE_CHECKING, Annotated

from sqlalchemy import String, Integer, ForeignKey, create_engine, DECIMAL, MetaData
from sqlalchemy.orm import (
    Mapped,
    DeclarativeBase,
    MappedAsDataclass,
    relationship,
    sessionmaker,
    mapped_column,
)

from config import settings

engine_1 = create_engine(str(settings.db_1_url), echo=True)
engine_2 = create_engine(str(settings.db_2_url), echo=True)

def get_metadata(schema: str):
    return MetaData(
        schema=schema,
        naming_convention={
            "ix": "ix_%(table_name)s_%(column_0_name)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",
        },
    )


class DB1Base(DeclarativeBase):
    __abstract__ = True
    metadata = get_metadata(settings.db_1_name)


class DB2Base(DeclarativeBase):
    __abstract__ = True
    metadata = get_metadata(settings.db_2_name)

IntPK = Annotated[int, mapped_column(primary_key=True, autoincrement=True)]

class User(MappedAsDataclass, DB2Base):
    __tablename__ = "users"

    id: Mapped[IntPK] = mapped_column(init=False)

    phones: Mapped[list["Phone"]] = relationship(init=False, back_populates="user")


class Phone(MappedAsDataclass, DB2Base):
    __tablename__ = "phones"

    id: Mapped[IntPK] = mapped_column(init=False)
    number: Mapped[Str50]

    user_id: Mapped[int] = mapped_column(Integer, ForeignKey(User.id), nullable=False)
    # new filed will be added soon
    # description: Mapped[str] = mapped_column(String(50), nullable=False)

    user: Mapped[User] = relationship(init=False, back_populates="phones")

When modifying the Phone model:

class Phone(...):
    user_id: Mapped[int] = mapped_column(ForeignKey(User.id))
+   description: Mapped[str] = mapped_column(String(50), nullable=False)

Alembic generates a migration that deletes and recreates the existing user_id foreign key even though the constraint name appears unchanged. The migration output shows:

INFO  [alembic.env] Migrating database test_2
INFO  [alembic.runtime.migration] Context impl MariaDBImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added column 'phones.description'
INFO  [alembic.autogenerate.compare] Detected removed foreign key (user_id)(id) on table phones
INFO  [alembic.autogenerate.compare] Detected added foreign key (user_id)(id) on table test_2.phones
Generating C:\Work\db-playground\alembic\versions\40133c68c820_2.py ...  done

Resulting in this upgrade script:

def upgrade_test_2() -> None:
    """Upgrade test_2 schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('phones', sa.Column('description', sa.String(length=50), nullable=False))
    op.drop_constraint('fk_phones_user_id_users', 'phones', type_='foreignkey')
    op.create_foreign_key(op.f('fk_phones_user_id_users'), 'phones', 'users', ['user_id'], ['id'], source_schema='test_2', referent_schema='test_2')
    # ### end Alembic commands ###

Key Observations:

1. The foreign key name (fk_phones_user_id_users) remains identical.
2. The migration removes and recreates the constraint.
3. The new constraint explicitly specifies source_schema and referent_schema.

Question:
Why does Alembic treat this as a constraint change requiring deletion/recreation instead of recognizing it as the same constraint? Is this related to schema-awareness in constraint comparisons, even when the constraint name remains unchanged?

[zzzeek]

hi -

it looks like the means by which Alembic is reflecting a foreign key from your database does not include "schema" information while the foreign key you indicate in your metadata includes a foreign key with a schema of "test_2". So these are two totally different FKs from Alembic's perspective.

Also this is not indicated but it looks from your object names that the driver in use is the DB2 driver? I dont know the state of the DB2 driver's reflection capabilities but correct round-tripping of schema information tends to be a more advanced use case.

if your DB2 connection works such that "test_2" is the default schema then you want to omit this schema name from your models.

https://docs.sqlalchemy.org/en/20/core/reflection.html#interaction-of-schema-qualified-reflection-with-the-default-schema

[lclbm]

./alembic.ini

databases = test_1, test_2
[test_1]
sqlalchemy.url = mariadb+mariadbconnector://root:password@localhost:3306/test_1
[test_2]
sqlalchemy.url = mariadb+mariadbconnector://root:password@localhost:3306/test_2

alembic/env.py

from database import DB1Base, DB2Base
target_metadata = {"test_1": DB1Base.metadata, "test_2": DB2Base.metadata}
...
def run_migrations_online() -> None:
    ...
    for name, rec in engines.items():
        logger.info("Migrating database %s" % name)
        context.configure(
            connection=rec["connection"],
            upgrade_token="%s_upgrades" % name,
            downgrade_token="%s_downgrades" % name,
            target_metadata=target_metadata.get(name),
        )
        context.run_migrations(engine_name=name)

./database.py

schema_1 = "test_1"
schema_2 = "test_2"
url = "mariadb+pymysql://root:password@localhost:3306/{}?charset=utf8mb4"

engine_1 = create_engine(url.format(schema_1), echo=True)
engine_2 = create_engine(url.format(schema_2), echo=True)


def get_metadata(schema: str):
    return MetaData(
        schema=schema,
        naming_convention={
            "ix": "ix_%(table_name)s_%(column_0_name)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",
        },
    )


class DB1Base(DeclarativeBase):
    __abstract__ = True
    metadata = get_metadata(schema_1)


class DB2Base(DeclarativeBase):
    __abstract__ = True
    metadata = get_metadata(schema_2)


IntPK = Annotated[int, mapped_column(primary_key=True, autoincrement=True)]


class User(MappedAsDataclass, DB1Base):
    __tablename__ = "users"

    id: Mapped[IntPK] = mapped_column(init=False)


class Phone(MappedAsDataclass, DB2Base):
    __tablename__ = "phones"

    id: Mapped[IntPK] = mapped_column(init=False)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey(User.id), nullable=False)

alembic command

alembic revision --autogenerate -m "t"
alembic upgrade head

In this setup, both SQLAlchemy and Alembic can handle cross-schema foreign keys. However, even when no table structure changes are made, running alembic revision --autogenerate results in all foreign keys being dropped and recreated.

INFO  [alembic.env] Migrating database test_1
INFO  [alembic.runtime.migration] Context impl MariaDBImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.env] Migrating database test_2
INFO  [alembic.runtime.migration] Context impl MariaDBImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected removed foreign key (user_id)(id) on table phones
INFO  [alembic.autogenerate.compare] Detected added foreign key (user_id)(id) on table test_2.phones
Generating D:\工作\test\alembic\versions\f34a7e2b8f2d_t3.py ...  done

def upgrade_test_2() -> None:
    """Upgrade test_2 schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint('fk_phones_user_id_users', 'phones', type_='foreignkey')
    op.create_foreign_key(op.f('fk_phones_user_id_users'), 'phones', 'users', ['user_id'], ['id'], source_schema='test_2', referent_schema='test_1')
    # ### end Alembic commands ###

[lclbm]

I've reviewed many discussions and issues related to this topic, but they all focus on PostgreSQL. I'm currently using MariaDB, which has been quite frustrating.

[zzzeek]

you have a cross-schema FK that is not going to be detected correctly by Alembic since you are running Alembic on both databases in schema-less mode. add an include_object routine to ignore this foreign key

def include_object(object, name, type_, reflected, compare_to):
    if name == "fk_phones_user_id_users":
        return False
    else:
        return True

context.configure(
    # ...
    include_object = include_object
)

[lclbm]

For MariaDB environments, do you have a solution that can both implement cross-database foreign keys in SQLAlchemy ORM and enable Alembic to correctly recognize and migrate them? I'm currently only able to achieve one of these at a time.
Thank u so much.

[zzzeek]

you would write these alembic migrations by hand for your cross-schema foreign keys.

if you want alembic autogenerate to automatically detect these cross-schema FKs (which IMO is not very important, just write the migrations for these by hand as needed, problem solved), you would need to run alembic with a single engine only, using schemas=True, along with models that are compatible with single engine connectivity (that is, rearrange your "schema" directives in your model metadata to correctly distinguish between the two databases in terms of a single database connection at once).