通過在郵件列表上@zzzeek:
我們做UPDATE..FROM但得到的VALUES東西在裏面需要一些額外的食譜,因爲我們沒有建在現在這個結構。也是它命名爲AS部分中的列的別名部分不是本地構建的。配方在這裏:https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PGValues。在某些時候,我向某人展示了AS的一部分,但那不在那裏...。我剛剛更新了它,它還需要幫助才能在UPDATE環境中工作。所以:
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:[email protected]/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))