beitreten Werte in sqlalchemy

Question

Ich möchte in sqlalchemy folgenden postgreSQL Abfrage auszuführen:

select c.* 
from comments c 
join (
    values 
    (1,1), 
    (3,2), 
    (2,3), 
    (4,4) 
) as x (id, ordering) on c.id = x.id 
order by x.ordering 

Ist es möglich, so etwas wie Liste von Listen oder einer Liste von Tupeln zu verbinden und nutzen sie Ordnung in sqlalchemy bieten?

Answer 1

from sqlalchemy import * 

from yourdbmodule import dbsession 


VALUES = ((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 it 
dbsession.query(Comments)\ 
    .join(temp_table, Comments.id == temp_table.c.id)\ 
    .order_by(temp_table.c.ordering)\ 
    .all() 

Answer 2

Siehe PGValues recipe für wie SQLAlchemy kompilieren die VALUES Klausel Gebrauch zu machen:

from sqlalchemy.ext.compiler import compiles 
from sqlalchemy.sql import column 
from sqlalchemy.sql.expression import FromClause 


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): 
     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 

Answer 3

Die Lösung von univerio funktioniert gut, es sei denn es einen Fehler aus, wenn Sie eine Spalte von den angegebenen Werten wählen:

>>> 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 

Dies liegt daran, die Werte in der bevölkerten Entfliehen und das JOIN Block

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

Die Lösung besteht darin, die Werte aus der FROM-Anweisung zu verbergen:

class values(FromClause): 
named_with_column = True 

def __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) 

Nun, es funktioniert:

>>> 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_ordering 
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