使用SQLAlchemy ORM从查询创建一个临时表

Question

使用SQLAlchemy ORM从查询创建一个临时表

9

我可以通过以下方式创建临时表：

session.execute("CREATE TABLE temptable SELECT existingtable.id, "
    "existingtable.column2 FROM existingtable WHERE existingtable.id<100000")

但是新表无法阅读，因为它说没有主键。 existingtable.id 是 existingtable 的主键，所以我期望临时表也会得到同样的对待。

然而，我更想找到一种 ORM 的方法来完成这个操作。给定：

temp_table = Table('temptable', metadata, 
    Column('id', Integer, primary_key=True),
    Column('column2', Integer),
    useexisting=True )
class TempTable(object):
    pass
mapper(TempTable, temp_table)
temp_table.create(bind=session.bind, checkfirst=True)
if session.query(TempTable).delete(): #make sure it's empty
    session.commit()

我该如何把existingtable中的一些选定内容填充到temp_table中，而不需要执行100000个session.query.add(TempTable(...))命令？或者是否有一种从查询中创建表格的方式，类似于上面的普通SQL版本？

- Paul

1

只是好奇..如果使用第一种方法创建临时表的定义，是否可以工作？例如：create table tablename (id primarykey, column1) Insert into tablename select id, column2 from existing table where... - AJP

@AJP 是的，我认为这就是答案。不幸的是，没有SA ORM等效的INSERT INTO，也没有简单的方法让我将相对复杂的查询编译成SQL语句。 - Paul

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- stephan · Accepted Answer

这并不是严格意义上的ORM，但为了创建表格，我会首先克隆表格结构（见下面的cloneTable示例）。要复制数据，我将使用InsertFromSelect示例。

编辑：自0.8.3版本以来，SqlAlchemy直接支持Insert.from_select()。因此，下面示例中的InsertFromSelect类和相应的访问者可以直接替换，并且不再需要。出于历史原因，我保留原始示例。

这是一个可行的示例：

from sqlalchemy import Table
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import UpdateBase

class InsertFromSelect(UpdateBase):
    def __init__(self, table, select):
        self.table = table
        self.select = select

@compiles(InsertFromSelect)
def visit_insert_from_select(element, compiler, **kw):
    return "INSERT INTO %s %s" % (
        compiler.process(element.table, asfrom=True),
        compiler.process(element.select)
    )

def cloneTable(name, table, metadata):
    cols = [c.copy() for c in table.columns]
    constraints = [c.copy() for c in table.constraints]
    return Table(name, metadata, *(cols + constraints))

# test data
from sqlalchemy import MetaData, Column, Integer
from sqlalchemy.engine import create_engine
e = create_engine('sqlite://')
m = MetaData(e)
t = Table('t', m, Column('id', Integer, primary_key=True),
          Column('number', Integer))
t.create()
e.execute(t.insert().values(id=1, number=3))
e.execute(t.insert().values(id=9, number=-3))

# create temp table
temp = cloneTable('temp', t, m)
temp.create()

# copy data
ins = InsertFromSelect(temp, t.select().where(t.c.id>5))
e.execute(ins)

# print result
for r in e.execute(temp.select()):
    print(r)