如何在SqlAlchemy+Alembic中引用基表字段定义Index？

背景 #

项目使用SqlAlchemy、Alembic和MyPy，定义了如下继承结构：

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),
    )

需要给emergency表添加基于基类InputBase中refresh_date和refresh_time字段的索引，执行迁移命令：

poetry run alembic --config operator_app/alembic.ini revision --autogenerate -m "refresh_col_indexes"

正确解决方案 #

有两种可靠的实现方式：

方式1：使用declared_attr动态定义__table_args__ #

通过sqlalchemy.ext.declarative.declared_attr装饰器，在类方法中以cls指代当前类，直接访问继承的字段：

from sqlalchemy.ext.declarative import declared_attr

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()

    @declared_attr
    def __table_args__(cls):
        return (
            PrimaryKeyConstraint("id", mssql_clustered=False),
            Index(
                "index_emergency_pii_patient_id_and_datetimes",
                cls.pii_patient_id,
                cls.attendance_start_date.desc(),
                cls.attendance_start_time.desc(),
            ),
            Index(
                "index_emergency_refresh_date_time",
                cls.refresh_date.desc(),
                cls.refresh_time.desc(),
            ),
        )

这种方式能让MyPy和Alembic都正确识别字段，同时支持排序方向的设置。

方式2：直接使用字段名称字符串 #

如果不需要指定排序方向，或者可以通过索引参数指定排序，直接用字段名的字符串形式：

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",
            "attendance_start_time",
            # 针对MSSQL设置降序（不同数据库语法可能有差异）
            mssql_descending=("attendance_start_date", "attendance_start_time")
        ),
        Index(
            "index_emergency_refresh_date_time",
            "refresh_date",
            "refresh_time",
            mssql_descending=("refresh_date", "refresh_time")
        ),
    )

之前方案失败的原因 #

• 直接写refresh_date.desc()：类级别定义中无法直接引用未在当前类显式定义的继承字段，MyPy无法识别该名称。
• 使用InputBase.refresh_date.desc()：InputBase是抽象基类，未绑定具体表，其字段属性不是有效的列对象，导致SqlAlchemy无法解析。
• 使用super()：super()只能在类的方法内部使用，类级别属性定义中不支持该语法。

内容的提问来源于stack exchange，提问作者Brondahl