SQL Server: Possible to tell Alembic to ignore sa.Identity(always=True)?

[abceleung]

I know that SQL Server does NOT support always=True in sa.Identity (because this is already the default behavior of SQL Server identity column), but I still added it because I try to be database agnostic.

The following script creates the model table in database with one identity column id:

# pip install SQLAlchemy alembic pyodbc

from sqlalchemy import (
    URL,
    Identity,
    create_engine,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class ModelBase(DeclarativeBase):
    pass


class Model(ModelBase):
    __tablename__ = "model"

    id: Mapped[int] = mapped_column(
        Identity(always=True, start=1, increment=1),
        primary_key=True,
    )


engine = create_engine(
    URL.create(
        "mssql+pyodbc",
        username="<redacted>",
        password="<redacted>",
        host="127.0.0.1",
        port=1433,
        database="<redacted>",
        query={
            "driver": "ODBC Driver 17 for SQL Server",
            "TrustServerCertificate": "yes",
        },
    )
)


ModelBase.metadata.create_all(engine)

Then I installed Alembic and setup autogenerate. After I ran alembic revision --autogenerate, I saw this section in the migration script:

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('model', 'id',
               existing_type=sa.INTEGER(),
               server_default=sa.Identity(always=True, start=1, increment=1),
               existing_nullable=False,
               autoincrement=True)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('model', 'id',
               existing_type=sa.INTEGER(),
               server_default=sa.Identity(always=False, start=1, increment=1),
               existing_nullable=False,
               autoincrement=True)
    # ### end Alembic commands ###

(Basically it trys to set sa.Identity from always=False to always=True)

If I run alembic upgrade head, an error occurs:

sqlalchemy.exc.CompileError: Adding, removing or modifying an "identity" construct, e.g. GENERATED AS IDENTITY, to or from an existing column is not supported in this dialect.

Is it possible to tell Alembic to ignore sa.Identity(always=True) when the database is SQL Server? Or is there an elegant way to handle this?

[CaselIT]

Hi,

You can control what alembic renders when using autogenerate with https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.environment.EnvironmentContext.configure.params.include_object

[abceleung]

Hi,

You can control what alembic renders when using autogenerate with https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.environment.EnvironmentContext.configure.params.include_object

I added the following to env.py:

import os

def include_object(
    object,
    name,
    type_,
    reflected,
    compare_to,
):
    if os.getenv("sqlalchemy_connection_string","").startswith("mssql") and type_ == "column":
        if object.info.get("alembic_skip_flag", "") == "sql_server":
            return False
    return True

Then I added include_object=include_object to all context.configure() calls below.
I also added info={"alembic_skip_flag": "sql_server"} to the definition of the identity column.

This solution is good enough for me, though I think it would omit any and all changes to the primary key column.
For completion's sake, is there a way to only ignore changes to the always=True flag in the Identity construct?

[CaselIT]

You could probably modify the identity in the reflected column. You can do that by doing this when the reflected attribute is true

[abceleung]

You could probably modify the identity in the reflected column. You can do that by doing this when the reflected attribute is true

Could you give me an example? I am reading the docs but haven't found a way to do it

[CaselIT]

I'm suggesting something like

def include_object(
    object,
    name,
    type_,
    reflected,
    compare_to,
):
    if os.getenv("sqlalchemy_connection_string","").startswith("mssql") and type_ == "column" and reflected:
        if object.info.get("alembic_normalize_identity", "") == "sql_server" and object.identity and object.identity.always is False:
            object.identity.always=True
    return True

I haven't tried it, but this should make the identity compare equally

[abceleung]

I'm suggesting something like

def include_object(
    object,
    name,
    type_,
    reflected,
    compare_to,
):
    if os.getenv("sqlalchemy_connection_string","").startswith("mssql") and type_ == "column" and reflected:
        if object.info.get("alembic_normalize_identity", "") == "sql_server" and object.identity and object.identity.always is False:
            object.identity.always=True
    return True

I haven't tried it, but this should make the identity compare equally

Thanks for your answer! object.identity.always is the thing I am missing.

[abceleung]

For future readers, you can use this include_object function to tell Alembic to ignore changes to Identity(always=) when your database is SQL Server:

import os 
from sqlalchemy.sql.schema import SchemaItem
import typing as ty

def include_object(
    # https://github.com/sqlalchemy/alembic/blob/main/alembic/runtime/environment.py#L66
    object: SchemaItem,
    name: ty.Optional[str],
    type_: ty.Literal[
        "schema",
        "table",
        "column",
        "index",
        "unique_constraint",
        "foreign_key_constraint",
    ],
    reflected: bool,
    compare_to: ty.Optional[SchemaItem],
) -> bool:
    if (
        # Check database is SQL Server
        os.getenv("sqlalchemy_connection_string", "").startswith("mssql")
        and type_ == "column"
    ):
        # If database is SQL Server AND column is Identity(always=True),
        # prevent Alembic from generating migration for identity.always
        if object.identity and object.identity.always is True:
            object.identity.always = False

    return True