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