使用從查詢創建臨時表SQLAlchemy的ORM

Question

我可以創建一個臨時表是這樣的：

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

但新表是不可讀的，因爲它說，它沒有主鍵。 existingtable.id是生存表的主要關鍵，所以我期望它在臨時表中得到相同的治療。

不過，我寧願找這樣反正一些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() 

我怎麼能填充temp_table與existingtable一些選擇的內容沒有做100000個session.query.add(TempTable(...))命令？或者有沒有一種方法可以通過類似於上面純SQL版本的查詢來創建表？

Answer 1

這不完全ORM，但最初創建表，我會克隆表結構（見下面的例子cloneTable）。爲了複製數據，我會使用InsertFromSelect example。

編輯：自從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