如何在 SqlAlchemy+Alembic 中在基表的列上定义 `Index()`

Question

我是一个Python新手。

我的项目正在使用 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()

我想向紧急表添加第二个索引，对基表提供的刷新列进行索引。

我希望通过在

Index()

设置中添加额外的

__table_args__

调用来实现此目的。

然后我想运行我的标准迁移创建/检查工具：

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

如何引用此声明中的刷新列？

---

当前尝试失败的猜测：

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

mypy 和 IDE 都说他们不知道

refresh_date

是什么。

error: Name "refresh_date" is not defined  [name-defined]

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

现在可以编译，但 alembic 命令不起作用：

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 说不：

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 说

Unsupported argument 2 for "super"

蒸馏器说

AttributeError: 'super' object has no attribute 'refresh_date'

Answer 1

您可以使用

sqlalchemy.orm.declared_attr

来实现此目的。您可以在

__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__(self):
        return (
            Index(
                "index_emergency_refresh_date_time",
                self.refresh_date.desc(),
                self.refresh_time.desc(),
            ),
        )

engine = create_engine("sqlite:///temp.sqlite", echo=True)
InputBase.metadata.create_all(engine)

这些是发出的查询。

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)