如何使用 SQLAlchemy 和 Postgres RETURNING 从一个表中删除行并插入另一个表？

Question

我想使用带有 Postgres 数据库的 SQLAlchemy 将行从一个表移动到另一个表（Stack Overflow 上还有其他关于移动数据的问题，但他们并不专注于为此使用 SQLAlchemy）。

方法是将DELETE 与RETURNING 一起使用，并将行插入到另一个表中。

我正在使用：SQLAlchemy 1.0.12、Postgres 9.4 和 Python 2.7.11。

设置表格

以下 SQL 创建表并插入一行数据：

create table example1 (
    id integer,
    value_a integer,
    value_b integer,
    CONSTRAINT example1_pkey PRIMARY KEY (id)
);
create table example2 (
    id integer,
    value_a integer,
    value_b integer,
    CONSTRAINT example2_pkey PRIMARY KEY (id)
);

insert into example1 values (18, 1, 9);

使用 SQLAlchemy 创建表

以下 SQLAlchemy 代码创建相同的表并插入一行数据：

from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class ExampleOne(Base):
    __tablename__ = 'example1'

    id = Column(Integer, primary_key=True)
    value_a = Column(Integer)
    value_b = Column(Integer)


class ExampleTwo(Base):
    __tablename__ = 'example2'

    id = Column(Integer, primary_key=True)
    value_a = Column(Integer)
    value_b = Column(Integer)


Base.metadata.create_all(session.bind)

with session.begin():
    session.add(ExampleOne(id=18, value_a=1, value_b=9))

我想实现的查询

这是我希望运行的 SQL 查询（独立运行）：

with output as (delete from example1 where value_a < 10 returning id, value_a)
insert into example2 (id, value_a, value_b)
select id, value_a, 3 from output;

到目前为止的 SQLAlchemy 查询

到目前为止我构建的查询是：

query = insert(ExampleTwo, inline=True).from_select(
    ['id', 'value_a', 'value_b'],
    select(
        ['id', 'value_a', literal(3)]
    ).where(
        select([
            'id', 'value_a',
        ]).select_from(
            delete(ExampleOne).where(
                ExampleOne.value_a < 10,
            ).returning(
                ExampleOne.id,
                ExampleOne.value_a,
            )
        )
    )
)

session.execute(query)

问题

错误是：

  File ".../lib/python2.7/site-packages/sqlalchemy/sql/selectable.py", line 41, in _interpret_as_from
    raise exc.ArgumentError("FROM expression expected")
sqlalchemy.exc.ArgumentError: FROM expression expected

问题似乎是 SQLAlchemy 无法将 DELETE ... RETURNING 查询识别为 INSERT 查询的 FROM 部分的有效表达式。

有没有办法让 SQLAlchemy 清楚这一点，或者是否有不同的方法可以在 SQLAlchemy 中创建给定的查询？

Answer 1

您需要将 delete 表达式转换为 CTE，因为您的原始 SQL 要求：

>>> output = delete(ExampleOne).where(
...     ExampleOne.value_a < 10,
... ).returning(
...     ExampleOne.id,
...     ExampleOne.value_a,
... ).cte('output')
>>> query = insert(ExampleTwo, inline=True).from_select(
...     ['id', 'value_a', 'value_b'],
...     select(
...         ['id', 'value_a', literal(3)]
...     ).select_from(output)
... )
>>> query.compile(engine)
WITH output AS 
(DELETE FROM example1 WHERE example1.value_a < %(value_a_1)s RETURNING example1.id, example1.value_a)
 INSERT INTO example2 (id, value_a, value_b) SELECT id, value_a, %(param_1)s AS anon_1 
FROM output

不幸的是，.cte 仅适用于当前未发布的 SQLAlchemy 1.1 中的 delete 表达式，因此您必须从源代码库安装 SQLAlchemy 才能使其正常工作：

pip install -e git+https://bitbucket.org/zzzeek/sqlalchemy#egg=sqlalchemy