如何利用SQLAlchemy将嵌套服务订单对象插入Oracle数据库？

一、优先利用关联关系自动处理（推荐） #

不用手动拆分操作，借助SQLAlchemy的关联关系+级联配置，可以一次性完成三张表的插入，同时保证事务一致性，代码简洁且不易出错。

1. 确保SQLAlchemy模型的关联关系配置正确 #

先确认表模型的外键和关联关系定义，重点配置relationship和cascade属性：

from sqlalchemy import Column, Integer, String, ForeignKey, Sequence
from sqlalchemy.orm import relationship, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class ServiceOrder(Base):
    __tablename__ = 'service_order'
    # Oracle序列生成自增主键
    service_order_id = Column(Integer, Sequence('service_order_seq'), primary_key=True)
    order_number = Column(String(50), unique=True)
    # 一对多关联订单项，级联操作包含新增/更新/删除
    service_order_items = relationship(
        "ServiceOrderItem", 
        back_populates="service_order", 
        cascade="all, delete-orphan"
    )

class ServiceOrderItem(Base):
    __tablename__ = 'service_order_item'
    item_id = Column(Integer, Sequence('order_item_seq'), primary_key=True)
    service_order_id = Column(Integer, ForeignKey('service_order.service_order_id'))
    # 一对一关联组件，uselist=False标记一对一，级联新增
    component = relationship(
        "Component", 
        back_populates="service_order_item", 
        uselist=False, 
        cascade="save-update"
    )
    service_order = relationship("ServiceOrder", back_populates="service_order_items")

class Component(Base):
    __tablename__ = 'component'
    component_id = Column(Integer, Sequence('component_seq'), primary_key=True)
    item_id = Column(Integer, ForeignKey('service_order_item.item_id'), unique=True)
    part_number = Column(String(50))
    service_order_item = relationship("ServiceOrderItem", back_populates="component")

2. Pydantic模型转SQLAlchemy实例 #

将嵌套的Pydantic模型转换为关联的SQLAlchemy对象：

from pydantic import BaseModel
from typing import List

# 示例Pydantic模型
class ComponentModel(BaseModel):
    part_number: str

class ServiceOrderItemModel(BaseModel):
    component: ComponentModel

class ServiceOrderModel(BaseModel):
    order_number: str
    service_order_items: List[ServiceOrderItemModel]

def pydantic_to_sqlalchemy(order_model: ServiceOrderModel) -> ServiceOrder:
    db_order = ServiceOrder(order_number=order_model.order_number)
    # 遍历订单项，关联组件
    for item_model in order_model.service_order_items:
        db_component = Component(part_number=item_model.component.part_number)
        db_item = ServiceOrderItem(component=db_component)
        db_order.service_order_items.append(db_item)
    return db_order

3. 事务内一次性插入 #

只需将根对象（ServiceOrder）添加到Session，commit时SQLAlchemy会自动按依赖顺序插入三张表，自动填充外键：

def insert_full_order(session: Session, order_model: ServiceOrderModel) -> int:
    try:
        db_order = pydantic_to_sqlalchemy(order_model)
        session.add(db_order)
        session.commit()
        # 直接从对象获取生成的主键ID
        return db_order.service_order_id
    except Exception as e:
        session.rollback()
        raise e

二、拆分操作的场景（特殊需求下） #

如果必须分步插入，可通过flush()在事务内提前获取生成的主键，避免数据不一致：

def split_insert_order(session: Session, order_model: ServiceOrderModel) -> int:
    try:
        # 1. 插入服务订单，flush获取自增ID（事务未提交）
        db_order = ServiceOrder(order_number=order_model.order_number)
        session.add(db_order)
        session.flush()
        order_id = db_order.service_order_id

        # 2. 插入订单项，flush获取item_id
        for item_model in order_model.service_order_items:
            db_item = ServiceOrderItem(service_order_id=order_id)
            session.add(db_item)
            session.flush()
            item_id = db_item.item_id

            # 3. 插入组件，用item_id作为外键
            db_component = Component(item_id=item_id, part_number=item_model.component.part_number)
            session.add(db_component)
        
        session.commit()
        return order_id
    except Exception as e:
        session.rollback()
        raise e

关键注意事项 #

• 事务一致性：无论哪种方式，都必须在事务内操作，避免部分插入成功的脏数据
• Oracle序列配置：自增主键需绑定Oracle序列，确保SQLAlchemy能正确获取生成的ID
• 级联规则：根据业务需求调整cascade参数，避免误删或漏更关联数据

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