Autogenerate keeps dropping and adding Foreign Key constraints

[drpoggi]

I'm having an issue where alembic --autogenerate keeps adding/dropping foreign key constraints in a MySQL 8 database despite no changes being made to the model and the foreign keys seeming to exist. I'm connecting using a DSN in the format mysql://{user}:{password}@{host}:{port}/myschema Here's the initial model:

roles = Table(
    "roles",
    meta,
    Column("id", Integer, primary_key=True, autoincrement=True),
)

api_tokens = Table(
    "api_tokens",
    meta,
    Column("id", Integer, primary_key=True, autoincrement=True),
)

token_roles = Table(
    "token_roles",
    meta,
    Column(
        "token_id",
        Integer,
        ForeignKey("api_tokens.id", ondelete="CASCADE"),
        primary_key=True,
    ),
    Column(
        "role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True
    ),
)

The autogenerate output is:

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('api_tokens',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.PrimaryKeyConstraint('id'),
    schema='myschema'
    )
    op.create_table('roles',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.PrimaryKeyConstraint('id'),
    schema='myschema'
    )
    op.create_table('token_roles',
    sa.Column('token_id', sa.Integer(), nullable=False),
    sa.Column('role_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['role_id'], ['myschema.roles.id'], ondelete='CASCADE'),
    sa.ForeignKeyConstraint(['token_id'], ['myschema.api_tokens.id'], ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('token_id', 'role_id'),
    schema='myschema'
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('token_roles', schema='myschema')
    op.drop_table('roles', schema='myschema')
    op.drop_table('api_tokens', schema='myschema')
    # ### end Alembic commands ###

After applying that and running the following, it looks like the foreign keys are created:

    from sqlalchemy import inspect

    with engine.connect() as conn:
        print(conn.exec_driver_sql("select DATABASE()").scalar())
        print(inspect(engine).get_foreign_keys("token_roles"))

[
    {
        "name": "token_roles_ibfk_1",
        "constrained_columns": ["role_id"],
        "referred_schema": None,
        "referred_table": "roles",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
    {
        "name": "token_roles_ibfk_2",
        "constrained_columns": ["token_id"],
        "referred_schema": None,
        "referred_table": "api_tokens",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
]

Now if I run autogenerate again I get the this revision:

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(op.f('token_roles_ibfk_1'), 'token_roles', type_='foreignkey')
    op.drop_constraint(op.f('token_roles_ibfk_2'), 'token_roles', type_='foreignkey')
    op.create_foreign_key(None, 'token_roles', 'roles', ['role_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    op.create_foreign_key(None, 'token_roles', 'api_tokens', ['token_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.create_foreign_key(op.f('token_roles_ibfk_2'), 'token_roles', 'api_tokens', ['token_id'], ['id'], ondelete='CASCADE')
    op.create_foreign_key(op.f('token_roles_ibfk_1'), 'token_roles', 'roles', ['role_id'], ['id'], ondelete='CASCADE')
    # ### end Alembic commands ###

And if I apply that and run that inspection:

[
    {
        "name": "token_roles_ibfk_1",
        "constrained_columns": ["role_id"],
        "referred_schema": None,
        "referred_table": "roles",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
    {
        "name": "token_roles_ibfk_2",
        "constrained_columns": ["token_id"],
        "referred_schema": None,
        "referred_table": "api_tokens",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
]

One more autogenerate/apply/inspection:

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(op.f('token_roles_ibfk_2'), 'token_roles', type_='foreignkey')
    op.drop_constraint(op.f('token_roles_ibfk_1'), 'token_roles', type_='foreignkey')
    op.create_foreign_key(None, 'token_roles', 'api_tokens', ['token_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    op.create_foreign_key(None, 'token_roles', 'roles', ['role_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.create_foreign_key(op.f('token_roles_ibfk_1'), 'token_roles', 'roles', ['role_id'], ['id'], ondelete='CASCADE')
    op.create_foreign_key(op.f('token_roles_ibfk_2'), 'token_roles', 'api_tokens', ['token_id'], ['id'], ondelete='CASCADE')
    # ### end Alembic commands ###

[
    {
        "name": "token_roles_ibfk_1",
        "constrained_columns": ["token_id"],
        "referred_schema": None,
        "referred_table": "api_tokens",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
    {
        "name": "token_roles_ibfk_2",
        "constrained_columns": ["role_id"],
        "referred_schema": None,
        "referred_table": "roles",
        "referred_columns": ["id"],
        "options": {"ondelete": "CASCADE"},
    },
]

I tried explicitly creating the foreign keys in the revision:

revision: str = '05316b3d5ad4'
down_revision: Union[str, Sequence[str], None] = '1a732f0eddb0'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('api_tokens',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.PrimaryKeyConstraint('id'),
    schema='myschema'
    )
    op.create_table('roles',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.PrimaryKeyConstraint('id'),
    schema='myschema'
    )
    op.create_table('token_roles',
    sa.Column('token_id', sa.Integer(), nullable=False),
    sa.Column('role_id', sa.Integer(), nullable=False),
    # sa.ForeignKeyConstraint(['role_id'], ['myschema.roles.id'], ondelete='CASCADE'),
    # sa.ForeignKeyConstraint(['token_id'], ['myschema.api_tokens.id'], ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('token_id', 'role_id'),
    schema='myschema'
    )
    op.create_foreign_key('token_roles_ibfk_1', 'token_roles', 'roles', ['role_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    op.create_foreign_key('token_roles_ibfk_2', 'token_roles', 'api_tokens', ['token_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    # ### end Alembic commands ###

However when I apply that and run autogenerate I still get the following:

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(op.f('token_roles_ibfk_2'), 'token_roles', type_='foreignkey')
    op.drop_constraint(op.f('token_roles_ibfk_1'), 'token_roles', type_='foreignkey')
    op.create_foreign_key(None, 'token_roles', 'roles', ['role_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    op.create_foreign_key(None, 'token_roles', 'api_tokens', ['token_id'], ['id'], source_schema='myschema', referent_schema='myschema', ondelete='CASCADE')
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.drop_constraint(None, 'token_roles', schema='myschema', type_='foreignkey')
    op.create_foreign_key(op.f('token_roles_ibfk_1'), 'token_roles', 'roles', ['role_id'], ['id'], ondelete='CASCADE')
    op.create_foreign_key(op.f('token_roles_ibfk_2'), 'token_roles', 'api_tokens', ['token_id'], ['id'], ondelete='CASCADE')
    # ### end Alembic commands ###

Any help is appreciated, it feels like I must be missing something very simple but I cannot figure out where I'm going wrong here. Thank you.

[zzzeek]

hi -

you have a schema mismatch. "myschema" is your default schema which means you need to make sure your models do not include "myschema" in their schema parameter, which should be left at None.

see the section https://docs.sqlalchemy.org/en/20/core/reflection.html#interaction-of-schema-qualified-reflection-with-the-default-schema for in depth detail here. this issue doesn't usually occur with MySQL since using the "schema" parameter is not that common on this backend.

[drpoggi]

I was including schema="myschema" in the MetaData object I was passing to my tables, removed that and now it's working, thank you for your help!