PostgreSQL 表继承和使用 SQLAlchemy 移动行答案

【问题标题】：PostgreSQL table inheritance and moving rows with SQLAlchemyPostgreSQL 表继承和使用 SQLAlchemy 移动行
【发布时间】：2016-05-15 19:25:15
【问题描述】：

在我的应用程序（python、postgres、sqlalchemy）中，我有一张大表tasks。该应用程序主要适用于tasks 中最近的 1k 行，经常选择和更新行。由于tasks 的大小，如此频繁的操作太慢了，所以我决定将此表拆分为两个taskstasks_all，其中表tasks 继承自tasks_all（postgresql 功能）。

因此，该应用可以非常快速地处理小表，并且当它对旧数据执行某些操作时，它可以处理大表，其中包括来自自身及其后继的所有行。

这是我的表格的简化类：

class TaskBase:
    def __init__(self, id, parent_id, data):
        self.id = id
        self.parent_id = parent_id
        self.data = data

    def __repr__(self):
        return '<Task: {} {}>'.format(self.id, self.data)


class Task(TaskBase, Base):
    __tablename__ = 'tasks'

    id = Column(INT, primary_key=True)
    parent_id = Column(INT, ForeignKey('tasks.id'))
    data = Column(TEXT)

    children = relationship("Task", backref=backref('parent', remote_side=[id]))


class TaskAll(TaskBase, Base):
    __tablename__ = 'tasks_all'

    id = Column(INT, primary_key=True)
    parent_id = Column(INT, ForeignKey('tasks_all.id'))
    data = Column(TEXT)

    children = relationship("TaskAll", backref=backref('parent', remote_side=[id]))

为了保持tasks 表较小，我需要时间将行从tasks 移动到tasks_all。有时我需要将它们移回去。以下代码是我尝试实现移动的方式。

def move_group(ids):
    query = session.query(TaskAll).filter(TaskAll.id in ids)
    db_session.s.execute(insert(Task).from_select((
        TaskAll.id,
        TaskAll.parent_id,
        TaskAll.data,
    ), query))
    print("insert from select is successful")
    query.delete()
    print("delete is successful")

但这不起作用，因为表tasks_all 是tasks 的父表，因此从tasks_all 中删除也会从tasks 中删除。生成的 SQL 查询是：

DELETE FROM tasks_all WHERE tasks_all.id IN (...);

但是要执行我需要的操作，查询应该是

DELETE FROM ONLY tasks_all WHERE tasks_all.id IN (...);

据我所知，SQLAlchemy 对ONLY 一无所知。我是否必须使用 sqla 跳过 ORM 部分的原始 SQL 查询，或者是否有更好的解决方案将行从一个表移动到另一个表？

【问题讨论】：

标签： python postgresql inheritance orm sqlalchemy

【解决方案1】：

我发现 SQLAlchemy 有一个提示机制，在对继承的表进行操作时会有所帮助。

def move_group(ids):
    query = session.query(TaskAll).filter(TaskAll.id in ids)
    db_session.s.execute(insert(Task).from_select((
        TaskAll.id,
        TaskAll.parent_id,
        TaskAll.data,
    ), query))
    print("insert from select is successful")
    query.delete().with_hint('ONLY', dialect_name='postgresql')
    print("delete is successful")

我知道这个问题是几年前提出的，但我希望它仍然可以帮助某人。

【讨论】：