加入 SQLAlchemy 中的 CTE

Question

我正在尝试制定一个 SQLAlchemy 查询，该查询使用 CTE 构建元组输入列表的类似 table 的结构，并将其与我的 tables（后端数据库是 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) 对 table 中的 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)
> ''

有办法吗？或者更好的方法来实现我的目标？

Answer 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()

加入 SQLAlchemy 中的 CTE

Join on a CTE in SQLAlchemy

python

postgresql

sqlalchemy

common-table-expression