crear una tabla temporal de una consulta utilizando sqlalchemy ORM

Question

que puede crear una tabla temporal de esta manera:

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

pero la nueva tabla es ilegible porque dice que no tiene clave primaria. existingtable.id es la clave principal de la tabla existente, por lo que esperaba obtener el mismo tratamiento en la tabla temporal.

Sin embargo, preferiría encontrar alguna forma ORM de hacer esto de todos modos. Teniendo en cuenta:

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

Cómo puede popular que temp_table con algunos contenidos seleccionados de existingtable sin hacer 100000 session.query.add(TempTable(...)) comandos? ¿O hay una forma de crear la tabla a partir de una consulta similar a la versión de SQL simple anterior?

Answer 1

No es exactamente ORM, pero para crear la tabla inicialmente, cloné la estructura de la tabla (consulte cloneTable en el ejemplo a continuación). Para copiar los datos, entonces usaría el InsertFromSelect example.

Editar:Desde la versión 0.8.3, sqlalchemy apoya Insert.from_select() fuera de la caja. Por lo tanto, la clase InsertFromSelect y el visitante respectivo en el ejemplo a continuación pueden reemplazarse directamente y ya no son necesarios. Dejo el ejemplo original sin cambios por razones históricas.

Aquí es un ejemplo de trabajo

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