SQLAlchemy如何基于字典配置循环动态创建不同列的ORM表

核心思路 #

将公共字段抽为抽象基类，通过Python内置的type()函数动态生成每个表对应的ORM类，无需逐个手动定义。

实现步骤 #

1. 定义公共字段抽象基类
   把所有表通用的字段封装到抽象基类中，避免重复编写：

from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class BaseDimensionTable(Base):
    __abstract__ = True  # 标记为抽象类，不会生成实际表
    id = Column(String(4), primary_key=True)
    shared_col1 = Column(String)
    shared_col2 = Column(String, nullable=False)

2. 准备表配置字典
   按照你给出的格式准备表名与对应维度字段的映射：

table_config = {
  "dimension_table1": ["dimension1", "dimension2"],
  "dimension_table2": ["dimension3"],
  # 可继续扩展其他表
}

3. 循环动态生成ORM类
   遍历配置字典，逐个生成表类并存储到映射字典中，方便后续调用：

# 存储生成的表类，key为表名，value为对应的ORM类
table_map = {}

for table_name, dim_fields in table_config.items():
    # 组装当前表的自定义属性
    attrs = {
        "__tablename__": table_name,
    }
    # 添加所有维度字段
    for field in dim_fields:
        attrs[field] = Column(String, nullable=False)
    # 动态生成类，继承自公共抽象基类
    table_cls = type(
        table_name,  # 类名，和表名保持一致即可
        (BaseDimensionTable,),  # 父类，继承公共字段
        attrs
    )
    table_map[table_name] = table_cls

使用示例 #

# 1. 创建所有表（和常规SQLAlchemy用法一致）
from sqlalchemy import create_engine
engine = create_engine("sqlite:///test.db")
Base.metadata.create_all(engine)

# 2. 调用指定表的ORM类
dim_table1 = table_map["dimension_table1"]
# 后续可直接用dim_table1进行增删改查操作，和手动定义的类用法完全一致

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