SQLAlchemy 动态/编程更新表定义

Posted 2023-03-25

【中文标题】SQLAlchemy 动态/编程更新表定义

【英文标题】：SQLAlchemy dynamically/programatically updating table definitions

【发布时间】：2021-11-06 00:57:00

【问题描述】：

我正在使用如下所示的 SQLAlchemy 模型：

class mytableclass(Base):
   __tablename__ = "table_name"

   __table__ = Table(
       __tablename__,
       Base.metadata,
      Column("Customer_id", String, primary_key=True),
      Column("Full_name", String),
      Column("Occupation", String),
 )

我有大约 50-60 个带有 sqlalchemy 表定义的 python 文件，我需要对它们进行一些更新。我有一个 csv，其中定义了类名和需要更新的内容，例如添加列、更改列类型或向列添加一些元数据。

现在，由于我正在处理如此大量的数据，我想知道如何以编程方式实现这一点，而不需要手动完成，因为将来会需要它。

一种方法是将 python 文件视为字符串并围绕它构建一些逻辑。但我有一种感觉，既然它是一个 python 数据结构，那么应该有一些“python”的方式来实现这一点。欢迎所有建议。

【参考方案1】：

您可以像下面的原型一样动态加载整个模型集。它不能处理很多边缘情况，例如将更多信息传递给表或建立关系。

我认为真正的问题是当您添加一列时，您必须更改表并可能操作/迁移现有数据以使其有意义。像 alembic 这样的东西可以尝试自动生成迁移，但大多数时候您必须自定义这些迁移或添加无法自动检测到的部分。因此，您可以从 csv 重新加载所有模型，然后运行手动编码迁移以更新数据库中的实际数据。与手动更改模型文件相比，我认为这真的不值得，因为您仍然需要编写迁移代码。

即使模型存储在 JSON 中并且您的迁移同时更改了 db 和 JSON，这也将是您一直在解决的一个巨大的自定义问题。 IE。这不处理没有相应类的关联表。会有很多边缘情况。不过这个想法很有趣。

最后有一些链接。

无论如何，这是我的想法：

#necessary sqlalchemy imports ...


Base = declarative_base()


engine = create_engine("sqlite://", echo=False)

# Constrain allowed types by filtering them through whitelist.
allowed_types = dict([(t.__name__, t) for t in (String, Integer,)])

# Extract this from CSV.
models = [
    'classname': 'Customer',
    'tablename': 'customers',
    'cols': [
        'name': 'customer_id', 'type': 'String', 'extra_kwargs': 'primary_key': True,
        'name': 'full_name', 'type': 'String',
        'name': 'occupation', 'type': 'String',
    ],
]

# Create class and table for each model and then shove the class
# into the globals()
for model in models:
    cols = [Column(col['name'], allowed_types[col['type']], **col.get('extra_kwargs', )) for col in model['cols']]
    globals()[model['classname']] = type(model['classname'], (Base,), 
        '__tablename__': model['tablename'],
        '__table__': Table(
            model['tablename'],
            Base.metadata,
            *cols),
    )

Base.metadata.create_all(engine)

with Session(engine) as session:
    customer = Customer(customer_id='some-guy', full_name='Some Guy', occupation='Some Job')
    session.add(customer)
    customers = session.query(Customer).all()
    assert len(customers) == 1
    assert customers[0].customer_id == 'some-guy'

alembic operations

dynamic class generation

type built-in

【讨论】：

嗨，Ian，感谢您的回答，这绝对可以帮助我走上正轨。不过有一些澄清：我没有完整的列列表，我只知道需要编辑的列，其余列需要以某种方式从类和 tabledefinition 中读取。最后，我如何将这个东西转储到 .py 文件中？还是这个 globals() 以某种方式实现它？我对此不熟悉，需要仔细阅读。

@Anton 我对此进行了更多思考，但这并不是您真正要问的。我认为操纵 python 代码本身将非常困难。这个解决方案只是从 JSON 样式的定义动态创建 Python sqlalchemy 类。因此，它会将所有模型文件替换为单个文件，该文件包含从 JSON 生成的所有模型。我不确定这是否会节省时间，因为实施尚未完成并且可能难以完全完成。