Create a table value with specific columns and values using a CTE statement

user_10288754 0
Viewed 772

I am trying to implement a PostgreSQL query in Python, using SQLAlchemy, but to no avail. My query is the following:

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

I tried many different approaches, but the main issue was that I could not create the "given" table with a CTE statement, being able to predefine the column names and give the required values.

This is a helpful thread on the VALUES clause in SQLAlchemy, but I still haven't managed to work the problem out.

Any insight on this issue would be more than welcome!

sqlalchemy postgresql python
1 Answers

So, after a bit of reading and tweaking I ended up with the following. First of all I had to add the code found here as an answer How to make use of bindparam() in a custom Compiled expression?. The code is an altered version of the code in 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

From there on the following code worked for me. I didn't have to use a CTE in the end. The crucial part was to declare column names for the temporary table, that wouldn't match my panel table column names, because it would lead to an "ambiguous column reference error".

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)
user_102887540
Related