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

Question

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'

Answer 1

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)