FastAPI中如何将含多字段header的Pydantic模型存入SQLAlchemy

在FastAPI中处理嵌套的header字段，有两种常见方案，可根据你的使用场景选择：

方案一：将Header的每个属性作为数据库单独列（适合需单独查询Header字段的场景） #

如果需要对Header里的属性做单独筛选、排序等操作，建议把每个属性拆成数据库表的单独列。

1. 修正SQLAlchemy模型 #

把Header中的每个属性作为可选列添加到MainActionTable（示例仅列4个，需补充剩余16个）：

class MainActionTable(Base):
    __tablename__ = 'MainActionTable'
    id = sql.Column(sql.Integer, primary_key=True, index=True)
    test_id = sql.Column(sql.String)
    status = sql.Column(sql.String)
    row_index = sql.Column(sql.Integer)  # 修正为int，匹配Pydantic模型类型
    test_type = sql.Column(sql.String)
    execute = sql.Column(sql.Boolean, nullable=True)  # 若存布尔值，后续需做字符串转布尔处理
    worksheet = sql.Column(sql.String)
    date_created = sql.Column(sql.DateTime, default=dt.datetime.utcnow)
    date_last_updated = sql.Column(sql.DateTime, default=dt.datetime.utcnow)
    
    # Header对应的字段，全部设为可选
    header_id = sql.Column(sql.String, nullable=True)
    header_execute = sql.Column(sql.String, nullable=True)
    header_task_name = sql.Column(sql.String, nullable=True)
    header_standard = sql.Column(sql.String, nullable=True)
    # 补充剩余16个Header属性，命名规则保持一致

2. 完善Pydantic模型 #

确保MainActionCreate包含所有必填字段：

class MainActionCreate(pydantic.BaseModel):
    worksheet: str
    row_index: int
    test_id: str
    test_type: str
    execute: str
    header: AdditionalData
    status: Optional[str] = None

3. 修正保存逻辑 #

将Header的属性拆分后赋值到数据库模型：

def create_action_point(db: orm.Session, action_create: MainActionCreate):
    # 提取Header数据并转换为数据库列对应的键名
    header_dict = action_create.header.dict()
    action_data = action_create.dict(exclude={"header"})
    
    # 合并Header字段到主数据
    action_data.update({
        "header_id": header_dict.get("ID"),
        "header_execute": header_dict.get("Execute"),
        "header_task_name": header_dict.get("TaskName"),
        "header_standard": header_dict.get("Standard"),
        # 补充剩余16个Header字段的映射
    })
    
    # 处理execute字段的类型转换（如果数据库存布尔值）
    action_data["execute"] = action_data["execute"].upper() == "ON"
    
    action_point = MainActionTable(**action_data)
    db.add(action_point)
    db.commit()
    db.refresh(action_point)
    return action_point

# 调用示例
action_tuple = data['action_case']
for x in action_tuple:
    # 处理原始数据中的header元组，转换为AdditionalData对象
    header_data = dict(zip(AdditionalData.model_fields.keys(), x['header']))
    create_data = MainActionCreate(
        worksheet=x['worksheet'],
        row_index=x['row_index'],
        test_id=x['test_id'],
        test_type=x['test_type'],
        execute=x['execute'],
        header=AdditionalData(**header_data),
        status=x['status']
    )
    create_action_point(db=db, action_create=create_data)

4. 读取时的映射 #

为让ORM模型自动转换为Pydantic模型，在MainAction中添加字段验证器组装Header数据：

class MainAction(pydantic.BaseModel):
    worksheet: str
    row_index: int
    test_id: str
    test_type: str
    execute: str
    header: AdditionalData
    status: Optional[str] = None

    class Config:
        orm_mode = True

    @pydantic.field_validator('header', mode='before')
    def assemble_header(cls, v, values):
        # 从ORM模型读取时，组装Header数据
        if isinstance(v, dict):
            return v
        return AdditionalData(
            ID=values.data.get('header_id'),
            Execute=values.data.get('header_execute'),
            TaskName=values.data.get('header_task_name'),
            Standard=values.data.get('header_standard'),
            # 补充剩余16个字段的映射
        )

方案二：用JSON类型存储整个Header（推荐，适合Header字段多、无需单独查询的场景） #

如果不需要对Header里的单个属性做数据库查询，直接把整个Header序列化为JSON存储，代码更简洁，维护成本更低。

1. 修正SQLAlchemy模型 #

添加一个header列，类型为sql.JSON：

class MainActionTable(Base):
    __tablename__ = 'MainActionTable'
    id = sql.Column(sql.Integer, primary_key=True, index=True)
    test_id = sql.Column(sql.String)
    status = sql.Column(sql.String)
    row_index = sql.Column(sql.Integer)  # 修正为int，匹配Pydantic类型
    test_type = sql.Column(sql.String)
    execute = sql.Column(sql.Boolean, nullable=True)  # 若存布尔值需转换
    worksheet = sql.Column(sql.String)
    header = sql.Column(sql.JSON)  # 用JSON类型存储整个Header数据
    date_created = sql.Column(sql.DateTime, default=dt.datetime.utcnow)
    date_last_updated = sql.Column(sql.DateTime, default=dt.datetime.utcnow)

2. 完善Pydantic模型 #

MainActionCreate直接继承MainAction即可：

class MainActionCreate(MainAction):
    pass

3. 简化保存逻辑 #

Pydantic会自动将AdditionalData转换为字典，SQLAlchemy的JSON列会直接存储：

def create_action_point(db: orm.Session, action_create: MainActionCreate):
    action_data = action_create.dict()
    # 处理execute字段的类型转换（如果数据库存布尔值）
    action_data["execute"] = action_data["execute"].upper() == "ON"
    action_point = MainActionTable(**action_data)
    db.add(action_point)
    db.commit()
    db.refresh(action_point)
    return action_point

# 调用示例
action_tuple = data['action_case']
for x in action_tuple:
    # 处理原始数据中的header元组，转换为AdditionalData对象
    header_data = dict(zip(AdditionalData.model_fields.keys(), x['header']))
    create_data = MainActionCreate(
        worksheet=x['worksheet'],
        row_index=x['row_index'],
        test_id=x['test_id'],
        test_type=x['test_type'],
        execute=x['execute'],
        header=AdditionalData(**header_data),
        status=x['status']
    )
    create_action_point(db=db, action_create=create_data)

4. 读取自动转换 #

因为Pydantic的orm_mode已开启，读取ORM模型时，JSON列的字典会自动转换为AdditionalData对象，无需额外处理。

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