使用 sqlalchemy orm 从查询创建临时表 [英] creating a temporary table from a query using sqlalchemy orm

问题描述

我可以这样创建临时表:

I can create a temporary table this way:

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

但是新表不可读，因为它说它没有主键.existingtable.id 是 exisitingtable 的主键，所以我希望它在临时表中得到相同的处理.

but the new table is unreadable because it says it has no primary key. existingtable.id is the primary key of exisitingtable, so I expected it to get the same treatment in the temp table.

然而，无论如何，我宁愿找到一些 ORM 方式来做到这一点.鉴于:

However, I would rather find some ORM way of doing this anyway. Given:

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

如何在不执行 100000 session.query.add(TempTable(...)) 的情况下使用 existingtable 的某些选定内容填充 temp_table 命令?或者有没有办法从类似于上面的纯 SQL 版本的查询中创建表?

How can I populate temp_table with some selected contents of existingtable without doing 100000 session.query.add(TempTable(...)) commands? Or is there a way of creating the table from a query similar to the plain SQL version above?

推荐答案

它不完全是 ORM，但为了最初创建表，我会克隆表结构(请参阅下面示例中的 cloneTable).为了复制数据，我将使用 InsertFromSelect 示例.

It's not exactly ORM, but to create the table initially, I'd clone the table structure (see cloneTable in the example below). For copying the data, I then would use the InsertFromSelect example.

从 0.8.3 版开始，SqlAlchemy 支持 Insert.from_select() 开箱即用.因此，下面示例中的 InsertFromSelect 类和相应的访问者可以直接替换，不再需要.由于历史原因，我保留原始示例不变.

Since version 0.8.3, SqlAlchemy supports Insert.from_select() out of the box. Hence the InsertFromSelect class and the respective visitor in the example below can be directly replaced and are no longer needed. I leave the original example unchanged for historic reasons.

这是一个工作示例

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)

这篇关于使用 sqlalchemy orm 从查询创建临时表的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！