9

我的程序中需要一个临时表。我已经看到这可以通过“映射器”语法以这种方式实现：

t = Table(
    't', metadata,
    Column('id', Integer, primary_key=True),
    # ...
    prefixes=['TEMPORARY'],
)

在这里看到

但是，我的整个代码都使用了声明式基础，这是我所理解的，我想坚持下去。有可能使用混合方法 ，但如果可能的话，我会避免它。

这是我的声明性类的简化版本：

import SQLAlchemy as alc
class Tempo(Base):
    """
    Class for temporary table used to process data coming from xlsx
    @param Base Declarative Base
    """

    # TODO: make it completely temporary

    __tablename__ = 'tempo'

    drw = alc.Column(alc.String)
    date = alc.Column(alc.Date)
    check_number = alc.Column(alc.Integer)

提前致谢！

用新问题编辑：

现在这个类看起来像这样：

import SQLAlchemy as alc

class Tempo(Base):
        """
        Class for temporary table used to process data coming from xlsx
        @param Base Declarative Base
        """

        # TODO: make it completely temporary

        __tablename__ = 'tempo'
        __table_args__ = {'prefixes': ['TEMPORARY']}

        drw = alc.Column(alc.String)
        date = alc.Column(alc.Date)
        check_number = alc.Column(alc.Integer)

当我尝试在此表中插入数据时，我收到以下错误消息：

sqlalchemy.exc.OperationalError: (OperationalError) no such table:
tempo u'INSERT INTO tempo (...) VALUES (?, ?, ?, ?, ?, ?, ?, ?)' (....)

似乎仅通过声明该表并不存在。我见过像 create_all() 这样的东西，它可能是解决这个问题的方法（在彻底解释的同时看到新想法是如何产生的，这很有趣）

再次，非常感谢您！

4

2 回答 2

5

可以使用__table_args__吗？请参阅https://docs.sqlalchemy.org/en/14/orm/declarative_tables.html#orm-declarative-table-configuration

class Tempo(Base):
    """
    Class for temporary table used to process data coming from xlsx
    @param Base Declarative Base
    """

    # TODO: make it completely temporary

    __tablename__ = 'tempo'
    __table_args__ = {'prefixes': ['TEMPORARY']}

    drw = alc.Column(alc.String)
    date = alc.Column(alc.Date)
    check_number = alc.Column(alc.Integer)
于 2012-11-19T14:32:30.410 回答
0

老问题，但如果有人想从现有的声明性表模型动态创建临时表，而不是让它始终成为模型/代码的一部分，您可以尝试以下方法。复制__table_args__有点棘手，因为它可以有多种格式，并且任何Index对象都需要重新创建，因此它们与旧表无关。

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 进行的轻微测试。另请注意，如果您使用非临时表和自动命名索引（即Column(..., index=True)）执行此操作，那么这可能与alembic.

于 2021-02-11T14:34:50.613 回答