python - 在连续的 pytests 中删除 sqlalchemy 列属性，在类声明之后添加

Question

要在数据库元数据中添加一列，在启动 alembic 迁移之前，我在列插入 python 脚本中执行以下操作：

class table(Base):
     __tablename__ = "my_table"
     id = Column(Integer, primary_key = True)
     name = Column(String)

col_name = "nickname"

#also assume  there's another class that has a foreign key that references table.id
related_class = list(inspect(table).relationships)[0].entity
related_tab = list(inspect(table).relationships)[0].entity.local_table

# I create two column objects, since the same one cannot be assigned to two tables
new_column = Column(col_name,type_ = new_type)
new_column2= Column(col_name,type_ = new_type)

table.__table__.append_column(new_column)
table.__mapper__.add_property(col_name,new_column)

#add the column to the related class/table
related_tab.append_column(new_column2)
related_class.add_property(col_name, new_column2)

#then I run my alembic auto revision and upgrade script

本质上，我创建了两个相同的列对象，然后将它们添加到表中，并作为映射器属性。

但是，我在测试数据库时遇到了问题。我的映射类（不是数据库）的状态被重用于下一个测试。但我希望在每次测试开始时都有一个干净的状态。

最后一个测试是向 Base.metadata 添加一个新列，然后创建一个自动生成的修订和升级。在测试结束时降级并不能解决问题。

这是具体情况。

我为每个测试类创建了一个新的引擎/数据库，但元数据/映射器（来自 Base 和类表）仍然包含来自前一组测试的额外列。作为我拆解的一部分，sqlite 数据库文件在测试集/类结束时被删除。

因此，下一组测试的 create_engine() 命令添加了我不想要的额外列。

使用 clear_mapper 不起作用，因为整个表都被删除了，并且无法在其他测试集中找到。

那么，作为拆解的一部分，如何从映射器中删除此列属性呢？

这是我发现几乎可以工作的东西

table.__mapper__.attrs=ImmutableProperties(
          dict(list(Ref_sheet.__mapper__.attrs.items())[:size-1]
              ))

table.__table__.columns = ColumnCollection(
        (list(table.__table__.columns.items())[:-1])).as_immutable()

如果我遍历列属性/键（），那么最新的列就消失了（在 pytest 类之间）。但是，如果我遍历表。mapper .all_ORM_descriptors()，应该删除的列仍然出现。

一个潜在的解决方案是为每组测试创建不同的类（即table1、table2）。但是，如果测试变得更大，这将无法扩展，并导致重复代码。

标签： pythonsqlalchemypytest

Answer 1

这只是删除类属性的问题，以及曾经通过首先检索它们而关联的列对象。但是，仍然存在 ORM 描述符 - 作为 InstrumentedAttributes。

但是，删除类属性和关联的列就足以让 Base.metadata.create_all() 创建没有删除列的表。需要做的只是在应用迁移脚本之前但在 create_all() 命令之后反映数据库中的表。

._columns 之前的下划线（我也假设属性）是列对象，然后将其转换为不可变列集合

del table.__mapper__._props['new_column']
#col.name is the name of the column key as it appears in the table
cols = [col for col in table.__table__._columns if col.name == 'new_column'][0]
table.__table__._columns.remove(cols)

Base.metadata.create_all()
table.__table__ = Table(table.__tablename__, Base.metadata, autoload_with = engine)

因此，在测试数据库操作以及单个列添加（例如使用 alembic）的 pytest 环境中，只需将上述代码包装在 if 语句中作为引擎设置固定装置的一部分。

if list(table.__table__.columns.keys())[-1] == 'new_column':
   del table.__mapper__._props['new_column']
   cols = [col for col in table.__table__._columns if col.name == 'new_column'][0]
   table.__table__._columns.remove(cols)
   #do the same for related tables

Base.metadata.create_all()
table.__table__ = Table(table.__tablename__, Base.metadata, autoload_with = engine)