在 SQLAlchemy 中加入 CTE答案

【问题标题】：Join on a CTE in SQLAlchemy在 SQLAlchemy 中加入 CTE
【发布时间】：2020-06-08 18:46:49
【问题描述】：

我正在尝试制定一个 SQLAlchemy 查询，该查询使用 CTE 来构建元组输入列表的类表结构，并将其与我的一个表（后端数据库是 Postgres）连接起来。从概念上讲，它看起来像：

WITH to_compare AS (
  SELECT * FROM (
    VALUES 
     (1, 'flimflam'),
     (2, 'fimblefamble'),
     (3, 'pigglywiggly'),
     (4, 'beepboop')
     -- repeat for a couple dozen or hundred rows
  ) AS t (field1, field2)
)
SELECT b.field1, b.field2, b.field3
FROM my_model b
JOIN to_compare c ON (c.field1 = b.field1) AND (c.field2 = b.field2)

目标是查看表中(field1, field2) 对的field3 是什么，以获取(field1, field2) 对的中等列表。

在 SQLAlchemy 中，我尝试这样做：

stmts = [
    sa.select(
        [
            sa.cast(sa.literal(field1), sa.Integer).label("field1"),
            sa.cast(sa.literal(field2), sa.Text).label("field2"),
        ]
    )
    if idx == 0
    else sa.select([sa.literal(field1), sa.literal(field2)])
    for idx, (field1, field2) in enumerate(list_of_tuples)
]
cte = sa.union_all(*stmts).cte(name="temporary_table")

already_in_db_query = db.session.query(MyModel)\
    .join(cte,
          cte.c.field1 == MyModel.field1,
          cte.c.field2 == MyModel.field2,
    ).all()

但似乎 CTE 和 JOIN 不能很好地配合使用：错误在 join 上，说：

sqlalchemy.exc.InvalidRequestError: Don't know how to join to ; please use an ON clause to more clearly establish the left side of this join

如果我尝试打印 cte，它看起来确实像一个非 SQL 实体：

$ from pprint import pformat
$ print(pformat(str(cte)), flush=True)
> ''

有没有办法做到这一点？或者更好的方式来实现我的目标？

【问题讨论】：

标签： python postgresql sqlalchemy common-table-expression

【解决方案1】：

Query.join() 的第二个参数在这种情况下应该是完整的 ON 子句，但是您将 3 个参数传递给 join()。使用and_() 组合谓词，就像在原始 SQL 中所做的那样：

already_in_db_query = db.session.query(MyModel)\
    .join(cte,
          and_(cte.c.field1 == MyModel.field1,
               cte.c.field2 == MyModel.field2),
    ).all()

【讨论】：