Присоединяйтесь к CTE в SQLAlchemy

Question

Я пытаюсь сформулировать запрос 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)

Цель состоит в том, чтобы увидеть, что field3 для пары (field1, field2) в таблице, если это так, для списка среднего размера из (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()