SQLAlchemy 临时表痛苦的解决方案？

Question

SQLAlchemy 的最大缺点似乎是在处理临时表时需要倒退几个步骤。例如，一个非常常见的用例是创建一个非常特定于一项任务的临时表，将一些数据放入其中，然后对其进行联接。

对于初学者来说，声明一个临时表是冗长且有限的。请注意，在此示例中，我必须对其进行编辑，因为我的类实际上继承了一个基类，所以我在这里给出的内容可能略有不正确。

@as_declarative(metaclass=MetaBase)
class MyTempTable(object):

    __tablename__ = "temp"
    __table_args__ = {'prefixes': ['TEMPORARY']}

    id = Column(Integer(), primary_key=True)
    person_id = Column(BigInteger())
    a_string = Column(String(100))

创建它是不直观的：

MyTempTable.__table__.create(session.bind)

我还必须记住明确删除它，除非我做一些创造性的事情来让它通过 ON COMMIT DROP 呈现：

MyTempTable.__table__.drop(session.bind)

此外，除非临时表完成“顶级”，否则我刚刚给出的内容甚至都不起作用。我还没有完全弄清楚这一点（因为不想花时间调查它为什么不起作用），但基本上我尝试使用 session.begin_nested() 在嵌套事务中以这种方式创建一个临时表，而你最终得到一个错误，说关系不存在。但是，我有几种情况，我在嵌套事务中创建一个临时表以进行单元测试，它们工作得很好。检查回显输出，似乎区别在于一个在 BEGIN 语句之前呈现，而另一个在它之后呈现。这是使用 Postgresql。

在嵌套事务中起作用并且坦率地说可以为您节省大量时间的是，只需键入该死的 sql 并使用 session.execute 执行它。

        session.execute(text(
            "CREATE TEMPORARY TABLE temp ("
            "  id SERIAL,"
            "  person_id BIGINT,"
            "  a_string TEXT"
            ") ON COMMIT DROP;"
        ))

当然，如果你这样做，你仍然需要一个相应的表模型来使用 ORM 功能，或者必须坚持使用原始 sql 查询，这首先违背了 SQLAlchemy 的目的。

我想知道我是否在这里遗漏了一些东西，或者是否有人提出了一个更优雅的解决方案。

Answer 1

我将 ORM 与 Core 一起使用。 ORM 保留用于更高级别的操作。对于大量数据和临时表，Core 更方便。示例：

temptbl_name = 'temp_del_dup_pk_{}'.format(datestamp)
temptbl = Table(temptbl_name, metadata, Column('col1', Integer, index=True),..., extend_existing=True)
temptbl.create(engine)

更新 这是一个可以动态生成临时表 ORM 定义的简单函数：

def temp_table(name, cols):
    args = dict(col1=Column(Integer, index=True),...)
    args['__tablename__'] = name
    args['__table_args__'] = dict(extend_existing=True)
    return type(name, (Base,), args)

镜像现有表的列会很有用：

def temp_table(name, base_table):
    args = {c.name:c.copy() for c in base_table.__table__.c}
    ...

Answer 2

我决定在this 答案的基础上进行构建，因为我想要一种更灵活的方式来从现有模型创建副本表，同时仍支持索引定义并与alembic* 配合使用。

我发现这种方法对于创建真正的临时表和创建将与主表交换的动态表都很有用。后者是如果定义不完全匹配，您可能会遇到更棘手的alembic 场景。

*根据我的特定使用模式

import time

from sqlalchemy.schema import CreateTable


def copy_table_args(model, **kwargs):
    """Try to copy existing __table_args__, override params with kwargs"""
    table_args = model.__table_args__

    if isinstance(table_args, tuple):
        new_args = []
        for arg in table_args:
            if isinstance(arg, dict):
                table_args_dict = arg.copy()
                table_args_dict.update(**kwargs)
                new_args.append(arg)
            elif isinstance(arg, sa.Index):
                index = sa.Index(
                    arg.name,
                    *[col for col in arg.columns.keys()],
                    unique=arg.unique,
                    **arg.kwargs,
                )
                new_args.append(index)
            else:
                # TODO: need to handle Constraints
                raise Exception(f"Unhandled table arg: {arg}")
        table_args = tuple(new_args)
    elif isinstance(table_args, dict):
        table_args = {
            k: (v.copy() if hasattr(v, "copy") else v) for k, v in table_args.items()
        }
        table_args.update(**kwargs)
    else:
        raise Exception(f"Unexpected __table_args__ type: {table_args}")

    return table_args


def copy_table_from_model(conn, model, **kwargs):
    model_name = model.__name__ + "Tmp"
    table_name = model.__table__.name + "_" + str(time.time()).replace(".", "_")
    table_args = copy_table_args(model, extend_existing=True)

    args = {c.name: c.copy() for c in model.__table__.c}
    args["__tablename__"] = table_name
    args["__table_args__"] = table_args

    copy_model = type(model_name, model.__bases__, args)
    print(str(CreateTable(copy_model.__table__)))
    copy_model.__table__.create(conn)
    return copy_model


def temp_table_from_model(conn, model, **kwargs):
    return copy_table_from_model(conn, model, prefixes=["TEMPORARY"])

注意：我没有添加逻辑来处理复制约束，这是针对 MySQL 进行的简单测试。