使用 CTE 语句创建具有特定列和值的表值

Question

我正在尝试使用 SQLAlchemy 在 Python 中实现 PostgreSQL 查询，但无济于事。我的查询如下：

with given ("id", "instance") as (
    values (1, 1), (108, 23), (203, 5)
)
select given."id" from given
left join panel on panel."id" = given."id" and panel.instance_id = given."instance"
where panel."id" is null

我尝试了许多不同的方法，但主要问题是我无法使用 CTE 语句创建“给定”表，无法预定义列名并提供所需的值。

这是VALUES clause in SQLAlchemy 上的一个有用线程，但我仍然没有设法解决问题。

欢迎任何有关此问题的见解！

Answer 1

所以，经过一番阅读和调整，我最终得到了以下内容。首先，我必须添加此处找到的代码作为答案How to make use of bindparam() in a custom Compiled expression?。该代码是VALUES clause in SQLAlchemy 中代码的修改版本。

from sqlalchemy import *
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import FromClause, ClauseElement
from sqlalchemy.dialects.postgresql import array
from sqlalchemy.types import NULLTYPE

from pp.core.db import panels, engine


class values(FromClause):
    named_with_column = True

    def __init__(self, columns, *args, **kw):
        self._column_args = columns
        self.list = args
        self.alias_name = self.name = kw.pop('alias_name', None)

    def _populate_column_collection(self):
        # self._columns.update((col.name, col) for col in self._column_args)
        for c in self._column_args:
            c._make_proxy(self, c.name)


@compiles(values)
def compile_values(clause, compiler, asfrom=False, **kw):
    def decide(value, column):
        add_type_hint = False
        if isinstance(value, array) and not value.clauses:  # for empty array literals
            add_type_hint = True

        if isinstance(value, ClauseElement):
            intermediate = compiler.process(value)
            if add_type_hint:
                intermediate += '::' + str(column.type)
            return intermediate

        elif value is None:
            return compiler.render_literal_value(
                value,
                NULLTYPE
            ) + '::' + str(column.type)
        else:
            return compiler.process(
                bindparam(
                    None,
                    value=compiler.render_literal_value(
                        value,
                        column.type
                    ).strip("'")
                )
            ) + '::' + str(column.type)

    columns = clause.columns
    v = "VALUES %s" % ", ".join(
        "(%s)" % ", ".join(
            decide(elem, column)
            for elem, column in zip(tup, columns))
        for tup in clause.list
    )
    if asfrom:
        if clause.alias_name:
            v = "(%s) AS %s (%s)" % (v, clause.alias_name, (", ".join(c.name for c in clause.columns)))
        else:
            v = "(%s)" % v
    return v

从那里开始，以下代码对我有用。我最终不必使用 CTE。关键部分是为临时表声明列名，这与我的面板表列名不匹配，因为它会导致“不明确的列引用错误”。

t1 = values(
    [
        column('given_id', Integer),
        column('given_instance', Integer)
    ],

    (1, 1),
    (108, 23),
    (203, 5),
    alias_name='given'
)

stmt = select(['given_id']). \
    select_from(t1.outerjoin(panels, and_(
        panels.c.id == t1.c.given_id,
        panels.c.instance_id == t1.c.given_instance
    ))). \
    where(
    panels.c.id.is_(None)
)

rs = engine.execute(stmt)