如何解析 JSON 模式文件并使用许多列约束动态创建新的 Python 类？

Posted 2023-03-16

【中文标题】如何解析 JSON 模式文件并使用许多列约束动态创建新的 Python 类？

【英文标题】：How to parse a JSON schema file and create new Python classes dynamically with many column constraints?

【发布时间】：2021-12-13 04:07:32

【问题描述】：

我将 SQLAlchemy 1.4 ORM 与 postgres13 和 Python 3.7 一起使用。

为清晰和完善而编辑：

要建立项目并对其进行测试，这 3 个文件运行良好：

base.py --> 设置 SQLAlchemy 引擎和数据库会话

models.py --> 一个用户类被定义了许多字段

inserts.py --> 创建 User 类的实例，添加并提交到数据库

如果models.py 已经定义了一个硬编码的类（例如用户类），这一切都很好。

我有一个定义数据库架构的schema.json 文件。该文件非常大，包含许多嵌套字典。 目标是解析 json 文件并使用给定的模式在models.py（或任何文件）中自动创建 Python 类。

schema.json 的一个例子：

  "groupings": 
    "imaging": 
      "owner":  "type": "uuid", "required": true, "index": true ,
      "tags":  "type": "text", "index": true 
      "filename":  "type": "text" ,
    ,

    "user": 
      "email":  "type": "text", "required": true, "unique": true ,
      "name":  "type": "text" ,
      "role": 
        "type": "text",
        "required": true,
        "values": [
          "admin",
          "customer",
        ],
        "index": true
      ,
      "date_last_logged":  "type": "timestamptz" 
    
  ,

  "auths": 
    "boilerplate": 
      "owner": ["read", "update", "delete"],
      "org_account": [],
      "customer": ["create", "read", "update", "delete"]
     ,

      "loggers": 
      "owner": [],
      "customer": []
    
  

模型的类需要通过解析 json 来动态创建，因为架构将来可能会发生变化，并且每次手动硬编码 100 多个类无法扩展。

我花了很多时间研究这个问题，但未能找到完全成功的解决方案。目前这就是我处理解析和动态表创建的方式。

我有这个函数create_class(table_data)，它传递了一个包含所有表名、列名、列约束的已解析字典。问题是，我无法弄清楚如何创建带有约束的表。目前，运行此代码会将表提交到数据库，但就列而言，它仅采用从 Base 继承的内容（自动生成的 PK ID）。

所有写入constraint_dict的列名和约束都被忽略。

#db_session.add(MyTableClass) 行被注释掉，因为它出现错误“sqlalchemy.orm.exc.UnmappedInstanceError: Class 'sqlalchemy.orm.decl_api.DeclarativeMeta' is not mapped; was a class (ma​​in .MyTableClass) 在需要实例的地方提供？"

我认为这一定与操作顺序有关——我在提交类本身之前创建了一个类的实例。我意识到这进一步混淆了事情，因为我实际上并没有调用 MyTableClass。

def create_class(table_data):

    constraint_dict = '__tablename__': 'myTableName'

    for k, v in table_data.items():

        if 'table' in k:
            constraint_dict['__tablename__'] = v

        else:
            constraint_dict[k] = f'Column(v)'


    MyTableClass = type('MyTableClass', (Base,), constraint_dict)
    Base.metadata.create_all(bind=engine)

    #db_session.add(MyTableClass)
    db_session.commit()
    db_session.close()

    return

我不太确定要查看什么来完成将所有列及其约束提交到数据库的最后一步。非常感谢任何帮助！

【问题讨论】：

这能回答你的问题吗？ Dynamic table creation SQLAlchemy ORM

【参考方案1】：

这并不能直接回答您的问题，而是提出了不同的策略。如果您希望 json 数据经常更改，您可以考虑创建一个带有 id 和 data 列的简单模型，本质上使用 postgres 作为 json 文档存储。

from sqlalchemy.dialects.postgresql import JSONB

class Schema(db.Model):
    id = db.Column(db.Integer(), nullable=False, primary_key=True, )
    data= db.Column(JSONB)

sqlalchemy：posgres dialect JSONB type

JSONB 数据类型比 posgres 中的 JSON 数据类型更可取，因为二进制表示的搜索效率更高，尽管 JSONB 的插入时间确实比 JSON 稍长。您可以在 posgres docs 中阅读有关 JSON 和 JSONB 数据类型之间区别的更多信息

这篇 SO 帖子解释了如何使用 SQLAlchemy 在 posgres 中执行 json 操作：sqlalchemy filter by json field

【讨论】：

谢谢！由于这个项目的限制，这个建议不会是一个选项。但是了解更多关于 JSONB 的信息很酷！