tạo ra một bảng tạm thời từ một truy vấn sử dụng SQLAlchemy ORM

Question

tôi có thể tạo ra một bảng tạm thời theo cách này:

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

nhưng bảng mới không thể đọc được vì nó nói nó không có khóa chính. existingtable.id là khóa chính của exisitingtable, vì vậy tôi mong đợi nó để có được điều trị tương tự trong bảng tạm thời.

Tuy nhiên, tôi thà tìm cách nào đó để thực hiện điều này. Đưa ra:

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

Làm thế nào tôi có thể cư temp_table với một số nội dung được lựa chọn của existingtable mà không làm 100000 session.query.add(TempTable(...)) lệnh? Hoặc là có một cách để tạo bảng từ một truy vấn tương tự như phiên bản SQL đơn giản ở trên?

Answer 1

Không chính xác ORM, nhưng để tạo bảng ban đầu, tôi sẽ sao chép cấu trúc bảng (xem cloneTable trong ví dụ bên dưới). Để sao chép dữ liệu, tôi sẽ sử dụng InsertFromSelect example.

Chỉnh sửa:Kể từ phiên bản 0.8.3, SqlAlchemy hỗ trợ Insert.from_select() ngoài hộp. Do đó, lớp InsertFromSelect và khách truy cập tương ứng trong ví dụ bên dưới có thể được thay thế trực tiếp và không còn cần thiết nữa. Tôi để nguyên ví dụ ban đầu không thay đổi vì lý do lịch sử.

Dưới đây là một ví dụ làm việc

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