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

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

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

我想向紧急情况表添加第二个索引,为基表提供的刷新列建立索引.

我希望通过在__table_args__设置中添加额外的Index()调用来实现这一点.

然后,我想运行我的标准迁移创建/判断工具: poetry run alembic --config operator_app/alembic.ini revision --autogenerate -m "refresh_col_indexes"

如何引用这个声明中的刷新列?


当前已失败的attem猜测:

    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" Alembic说AttributeError: 'super' object has no attribute 'refresh_date'

推荐答案

你可以用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__(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)

这些是发出的查询.

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)

Python相关问答推荐

Polars比较了两个预设-有没有方法在第一次不匹配时立即失败

根据不同列的值在收件箱中移动数据

Deliveryter Notebook -无法在for循环中更新matplotlib情节(保留之前的情节),也无法使用动画子功能对情节进行动画

删除最后一个pip安装的包

发生异常:TclMessage命令名称无效.!listbox"

如何在polars(pythonapi)中解构嵌套 struct ?

如何在类和classy-fastapi -fastapi- followup中使用FastAPI创建路由

Pandas—合并数据帧,在公共列上保留非空值,在另一列上保留平均值

python中的解释会在后台调用函数吗?

启用/禁用shiny 的自动重新加载

Tkinter菜单自发添加额外项目

网格基于1.Y轴与2.x轴显示在matplotlib中

为什么后跟inplace方法的`.rename(Columns={';b';:';b';},Copy=False)`没有更新原始数据帧?

浏览超过10k页获取数据,解析:欧洲搜索服务:从欧盟站点收集机会的微小刮刀&

Python日志(log)库如何有效地获取lineno和funcName?

类型对象';敌人';没有属性';损害';

从列表中分离数据的最佳方式

时长超过24小时如何从Excel导入时长数据

使用xlsxWriter在EXCEL中为数据帧的各行上色

大Pandas 每月重新抽样200万只和300万只