如何使用SQLAlchemy Postgres ORM声明基类从字典动态创建带约束的表

1. 你传递给动态类的列定义是字符串格式（例如constraint_dict[k] = f'= Column({v})'），但SQLAlchemy声明式基类要求类属性必须是实际的Column实例，无法识别字符串作为列定义，所以只有继承自Base的主键id能被正常识别。
2. 操作逻辑混淆了三个不同层级的能力边界：
   • db_session.add仅用于添加数据行实例，不能用来注册模型类
   • 手动调用Table构造属于SQLAlchemy Core层操作，和声明式ORM的自动表注册逻辑冲突
   • 不需要手动给类指定__tablename__，你在Base中已经通过declared_attr配置了自动将类名转小写作为表名的逻辑。

1. 定义类型与约束映射 #

先把JSON中定义的字段类型、约束规则映射为SQLAlchemy可识别的实际对象：

from sqlalchemy import Column, Text, Integer, DateTime, CheckConstraint
from sqlalchemy.dialects.postgresql import UUID
from base import Base, engine
import uuid

# JSON字段类型映射到SQLAlchemy类型对象
TYPE_MAPPING = {
    "uuid": UUID(as_uuid=True),
    "text": Text,
    "bigint": Integer,
    "timestamptz": DateTime(timezone=True)
}

def convert_snake_to_camel(name):
    return ''.join(x.capitalize() or '_' for x in name.split('_'))

2. 重写动态类创建逻辑 #

直接生成Column实例存入类属性字典，不要生成字符串：

def create_class(table_name, column_configs):
    # 类属性字典，存储Column实例和表配置
    class_attrs = {}
    table_args = []

    for col_name, col_conf in column_configs.items():
        # 解析列参数
        col_kwargs = {
            "nullable": not col_conf.get("required", False),
            "index": col_conf.get("index", False),
            "unique": col_conf.get("unique", False)
        }
        # uuid类型默认加生成函数
        if col_conf["type"] == "uuid":
            col_kwargs["default"] = uuid.uuid4
        # 枚举值加校验约束
        if "values" in col_conf:
            table_args.append(CheckConstraint(f"{col_name} in {tuple(col_conf['values'])}"))
        
        # 生成实际的Column实例
        class_attrs[col_name] = Column(TYPE_MAPPING[col_conf["type"]], **col_kwargs)
    
    # 有自定义约束的话加入类属性
    if table_args:
        class_attrs["__table_args__"] = tuple(table_args)
    
    # 生成ORM模型类，自动继承Base的主键和表名规则
    table_cls = type(convert_snake_to_camel(table_name), (Base,), class_attrs)
    return table_cls

3. 批量生成类并建表 #

解析完schema.json后遍历所有表配置生成类，最后统一调用建表接口即可：

import json

if __name__ == "__main__":
    # 加载schema配置
    with open("schema.json", "r", encoding="utf-8") as f:
        schema = json.load(f)
    
    # 批量生成所有表的ORM类
    for table_name, column_configs in schema["groupings"].items():
        create_class(table_name, column_configs)
    
    # 统一建表，checkfirst=True避免重复创建报错
    Base.metadata.create_all(bind=engine, checkfirst=True)

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