join values in sqlalchemy join values in sqlalchemy postgresql postgresql

join values in sqlalchemy


from sqlalchemy import *from yourdbmodule import dbsessionVALUES = ((1, 1), (3, 2), (2, 3), (4, 4))temp_table = Table(    'temp_table', MetaData(),    Column('id', INT, primary_key=True),    Column('ordering', INT),    prefixes=['TEMPORARY'])temp_table.create(bind=dbsession.bind, checkfirst=True)dbsession.execute(temp_table.insert().values(VALUES))# Now you can query itdbsession.query(Comments)\    .join(temp_table, Comments.id == temp_table.c.id)\    .order_by(temp_table.c.ordering)\    .all()


See the PGValues recipe for how to make SQLAlchemy compile the VALUES clause:

from sqlalchemy.ext.compiler import compilesfrom sqlalchemy.sql import columnfrom sqlalchemy.sql.expression import FromClauseclass 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):        for c in self._column_args:            c._make_proxy(self)@compiles(values)def compile_values(element, compiler, asfrom=False, **kw):    columns = element.columns    v = "VALUES %s" % ", ".join(        "(%s)" % ", ".join(                compiler.render_literal_value(elem, column.type)                for elem, column in zip(tup, columns))        for tup in element.list    )    if asfrom:        if element.alias_name:            v = "(%s) AS %s (%s)" % (v, element.alias_name, (", ".join(c.name for c in element.columns)))        else:            v = "(%s)" % v    return v>>> x = values([column("id", Integer), column("ordering", Integer)], (1, 1), (3, 2), (2, 3), (4, 4), alias_name="x")>>> q = session.query(Comment).join(x, Comment.id == x.c.id).order_by(x.c.ordering)>>> print(q)SELECT comments.id AS comments_id FROM comments JOIN (VALUES (1, 1), (3, 2), (2, 3), (4, 4)) AS x (id, ordering) ON comments.id = x.id ORDER BY x.ordering


The solution of univerio works great, except it throws an error if you select a column from the given values:

>>> q = session.query(Comment, c.x.id).join(x, Comment.id == x.c.id).order_by(x.c.ordering)>>> q.all()ProgrammingError (ProgrammingError) table name "x" specified more than once

This is because the VALUES get populated into the FROM and the JOIN block

>>> print(q)SELECT comments.id AS comments_id, x.ordering as x_orderingFROM (VALUES (1, 1), (3, 2), (2, 3), (4, 4)) AS x (id, ordering), commentsJOIN (VALUES (1, 1), (3, 2), (2, 3), (4, 4)) AS x (id, ordering) ON comments.id = x.id ORDER BY x.ordering

The solution is to hide the VALUES from the FROM statement:

class values(FromClause):named_with_column = Truedef __init__(self, columns, *args, **kw):    self._column_args = columns    self.list = args    self._hide_froms.append(self)  # This line fixes the above error    self.alias_name = self.name = kw.pop('alias_name', None)def _populate_column_collection(self):    for c in self._column_args:        c._make_proxy(self)

Now it works fine:

>>> q = session.query(Comment, c.x.id).join(x, Comment.id == x.c.id).order_by(x.c.ordering)>>> print(q)SELECT comments.id AS comments_id, x.ordering as x_orderingFROM comments JOIN (VALUES (1, 1), (3, 2), (2, 3), (4, 4)) AS x (id, ordering) ON comments.id = x.id ORDER BY x.ordering