pythoninheritanceindexingsqlalchemyalembic

How to define an `Index()` in SqlAlchemy+Alembic, on a column from a base table


I am a python novice.

My project is using SqlAlchemy, Alembic and MyPy.

I have a pair of parent-child classes defined like this (a bunch of detail elided):

class RawEmergency(InputBase, RawTables):
    __tablename__ = "emergency"

    id: Mapped[UNIQUEIDENTIFIER] = mapped_column(
        UNIQUEIDENTIFIER(), primary_key=True, autoincrement=False
    )

    attendance_id: Mapped[str | None] = guid_column()
    admitted_spell_id: Mapped[str | None] = guid_column()


    __table_args__ = (
        PrimaryKeyConstraint("id", mssql_clustered=False),
        Index(
            "index_emergency_pii_patient_id_and_datetimes",
            pii_patient_id,
            attendance_start_date.desc(),
            attendance_start_time.desc(),
        ),
    )

class InputBase(DeclarativeBase):
    metadata = MetaData(schema="raw")

    refresh_date: Mapped[str] = date_str_column()
    refresh_time: Mapped[str] = time_str_column()

class RawTables(object):
    id: Mapped[UNIQUEIDENTIFIER] = mapped_column(
        UNIQUEIDENTIFIER(), primary_key=True, autoincrement=False
    )

    __table_args__: typing.Any = (
        PrimaryKeyConstraint(name="id", mssql_clustered=False),
    )

I want to add a 2nd index to the Emergency table, indexing the refresh columns provided by the base table.

I expect to do so by adding an additional Index() call into the __table_args__ setup.

Then I want to run my standard migration creation/checking tool: poetry run alembic --config operator_app/alembic.ini revision --autogenerate -m "refresh_col_indexes"

How do I reference the refresh columns in this declaration?


Current attemguesses that have failed:

    Index(
        "index_emergency_refresh_date_time",
        refresh_date.desc(),
        refresh_time.desc(),
    ),

mypy and the IDE both say they don't know what refresh_date is. error: Name "refresh_date" is not defined [name-defined]

    Index(
        "index_emergency_refresh_date_time",
        InputBase.refresh_date.desc(),
        InputBase.refresh_time.desc(),
    ),

compiles now, but the alembic command doesn't work: sqlalchemy.exc.ArgumentError: Can't add unnamed column to column collection full error below

    Index(
        "index_emergency_refresh_date_time",
        super().refresh_date.desc(),
        super().refresh_time.desc(),
    ),

Mypy/IDE say no: error: "super()" outside of a method is not supported

    Index(
        "index_emergency_refresh_date_time",
        super(InputBase, self).refresh_date.desc(),
        super(InputBase, self).refresh_time.desc(),
    ),

self is not defined

    Index(
        "index_emergency_refresh_date_time",
        super(InputBase, None).refresh_date.desc(),
        super(InputBase, None).refresh_time.desc(),
    ),

mypy says Unsupported argument 2 for "super" and alembic says AttributeError: 'super' object has no attribute 'refresh_date'


Solution

  • You can use sqlalchemy.orm.declared_attr for this. You can add any number of index you want under __table_args__

    from sqlalchemy import create_engine, Index
    from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, declared_attr
    
    
    class InputBase(DeclarativeBase):
        refresh_date: Mapped[str]
        refresh_time: Mapped[str]
    
    
    class RawEmergency(InputBase):
        __tablename__ = "emergency"
    
        id: Mapped[int] = mapped_column(primary_key=True, autoincrement=False)
    
        attendance_id: Mapped[str | None]
        admitted_spell_id: Mapped[str | None]
    
        @declared_attr
        def __table_args__(cls):
            return (
                Index(
                    "index_emergency_refresh_date_time",
                    cls.refresh_date.desc(),
                    cls.refresh_time.desc(),
                ),
            )
    
    engine = create_engine("sqlite:///temp.sqlite", echo=True)
    InputBase.metadata.create_all(engine)
    

    These are the queries emitted.

    CREATE TABLE emergency (
            id INTEGER NOT NULL, 
            attendance_id VARCHAR, 
            admitted_spell_id VARCHAR, 
            refresh_date VARCHAR NOT NULL, 
            refresh_time VARCHAR NOT NULL, 
            PRIMARY KEY (id)
    )
    
    sqlalchemy.engine.Engine CREATE INDEX index_emergency_refresh_date_time ON emergency (refresh_date DESC, refresh_time DESC)