I want to write a query like so with SQLAlchemy:
UPDATE mytable
SET
mytext = myvalues.mytext,
myint = myvalues.myint
FROM (
VALUES
(1, 'textA', 99),
(2, 'textB', 88),
...
) AS myvalues (mykey, mytext, myint)
WHERE mytable.mykey = myvalues.mykey
Is this kind of thing supported natively in the ORM? Or will I need to use session.execute() to run raw SQL?
Via @zzzeek on the mailing list:
We do UPDATE..FROM but getting the VALUES thing in there requires some extra recipes as we don’t have that structure built in right now. also the alias part of it where it names out the columns in the AS portion is not natively built in either. The recipe is here: https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PGValues. At some point I showed someone the AS part, but that’s not in there….OK I’ve just updated it, it also needed help to work in the UPDATE context. So:
from sqlalchemy.ext.compiler import compiles
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
if __name__ == '__main__':
from sqlalchemy import MetaData, create_engine, String, Integer, Table, Column
from sqlalchemy.sql import column
from sqlalchemy.orm import Session, mapper
m1 = MetaData()
class T(object):
pass
t1 = Table('mytable', m1, Column('mykey', Integer, primary_key=True),
Column('mytext', String),
Column('myint', Integer))
mapper(T, t1)
t2 = values(
[
column('mykey', Integer),
column('mytext', String),
column('myint', Integer)
],
(1, 'textA', 99),
(2, 'textB', 88),
alias_name='myvalues'
)
e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
m1.create_all(e)
sess = Session(e)
sess.query(T).filter(T.mykey==t2.c.mykey).\
update(dict(mytext=t2.c.mytext, myint=t2.c.myint))