如何解析JSON Schema动态创建带列约束的SQLAlchemy ORM类

核心错误点 #

你遇到的两个问题根源非常明确：

1. 错误使用字符串代替Column对象：你现有代码中给constraint_dict赋值的是f'Column({v})'格式的字符串，SQLAlchemy的声明式基类只会识别类属性中真实的Column实例，字符串会被当成普通类属性忽略，自然不会生成对应表字段。
2. 错误将类传入session.add：db_session.add()方法仅接收模型类的实例对象（比如MyTableClass()），传入类本身就会抛出UnmappedInstanceError。建表属于元数据操作，完全不需要操作session，仅调用Base.metadata.create_all即可完成。

完整实现方案 #

步骤1：导入依赖与预配置类型映射 #

首先建立JSON字段类型到SQLAlchemy字段类型的映射，同时处理各类约束逻辑：

from sqlalchemy import Column, UUID, Text, TIMESTAMP, CheckConstraint, Enum
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
# 导入你自己的Base、engine、db_session
from base import Base, engine, db_session
import json
import uuid

# JSON类型到SQLAlchemy类型的映射
TYPE_MAPPING = {
    "uuid": UUID(as_uuid=True),
    "text": Text,
    "timestamptz": TIMESTAMP(timezone=True)
}

# 存储所有动态生成的模型类，后续操作数据时可以直接取用
dynamic_models = {}

步骤2：改造动态生成类的函数 #

def create_model_class(table_name: str, fields: dict) -> type:
    # 构造类属性字典
    class_attrs = {
        "__tablename__": table_name,
        # 如果你的Base类没有预定义主键，需要在这里主动添加主键字段
        "id": Column(PG_UUID, primary_key=True, default=uuid.uuid4)
    }

    for field_name, field_config in fields.items():
        # 构造Column参数
        col_type = TYPE_MAPPING[field_config["type"]]
        col_kwargs = {
            "nullable": not field_config.get("required", False),
            "index": field_config.get("index", False),
            "unique": field_config.get("unique", False)
        }

        # 处理枚举值约束
        if "values" in field_config:
            enum_name = f"{table_name}_{field_name}_enum"
            col_type = Enum(*field_config["values"], name=enum_name)
            # 可选：额外加CheckConstraint做双重校验
            class_attrs[f"_{field_name}_check"] = CheckConstraint(
                f"{field_name} IN {tuple(field_config['values'])}"
            )

        # 生成真实的Column实例，而非字符串
        class_attrs[field_name] = Column(col_type, **col_kwargs)

    # 动态生成模型类，类名用表名的大驼峰格式，比如user -> User
    model_class = type(table_name.capitalize(), (Base,), class_attrs)
    dynamic_models[table_name] = model_class
    return model_class

步骤3：解析schema.json并批量生成模型、建表 #

def generate_models_from_schema(schema_path: str):
    with open(schema_path, "r", encoding="utf-8") as f:
        schema = json.load(f)
    
    # 仅处理groupings下的表定义，auths为权限配置不需要生成表
    for table_name, fields in schema["groupings"].items():
        create_model_class(table_name, fields)
    
    # 所有类生成完成后统一建表，不需要每个类生成就调用一次
    Base.metadata.create_all(bind=engine)

步骤4：使用动态生成的模型操作数据 #

if __name__ == "__main__":
    # 生成所有模型与表
    generate_models_from_schema("schema.json")

    # 插入数据示例：插入user表数据
    User = dynamic_models["user"]
    new_user = User(
        email="test@example.com",
        name="测试用户",
        role="admin"
    )
    db_session.add(new_user)
    db_session.commit()

可选方案：静态生成models.py文件 #

如果你不想每次程序启动都动态生成类，也可以直接输出静态的models.py文件，后续schema变更时只要重新跑一次生成脚本即可：

1. 预先写好models.py的模板，包含导入语句、Base类引入
2. 把解析出来的表、字段信息拼写成标准的模型类代码字符串
3. 写入到models.py文件即可，不需要运行时动态生成类。

内容的提问来源于stack exchange，提问作者alphaomega