SQLAlchemy父子表插入回滚实现及异常问题排查

问题场景 #

使用SQLAlchemy实现Order（父表）和OrderDetails（子表）的事务插入，同时插入父表数据和多条子表数据时遇到以下问题：

• 调用db.bulk_insert_mappings(order, order_details)时触发错误：TypeError: 'OrderDetails' object is not iterable
• 调用db.add(order)时，API传入两条子表数据，但最终仅插入一条

相关代码 #

Pydantic模型及业务代码 #

class OrderDetailModel(BaseModel):
    prod_id: int
    price_id: int
    amount: int

class OrderModel(BaseModel):
    user_id: int
    order_detail: List[OrderDetailModel] = []

    def create_order(self, db: Session):
        db.begin()
        order_details: OrderDetails = []
        od: OrderDetails = OrderDetails()
        for order in self.order_detail:
            od.prod_id = order.prod_id
            od.price_id = order.price_id
            od.quantity = 1
            od.amount = order.amount
            order_details.append(od)
        print('Order_details list size >>>>>>>>>>>>>>>>>>>>', len(order_details))

        order = Order(unique_order_id=''.join(random.choices(string.ascii_lowercase, k=10)),
                                 order_date=datetime.utcnow().date(),
                                 checkout_date=datetime.utcnow().date(),
                                 user_id=self.user_id,
                                 order_status="FINISHED",
                                 order_amount=100,
                                 order_details=order_details)

        db.bulk_insert_mappings(order, order_details)    # 触发TypeError
        # db.add(order)  # 仅插入一条子表数据
        db.commit()

        return "success"

数据表结构 #

class Order(Model):
    __tablename__="marketplace_orders"
    id = sa.Column(BIGINT, primary_key=True, autoincrement=True)
    unique_order_id = sa.Column(sa.String(),  unique=True, nullable=False)
    order_date= sa.Column(types.DateTime, nullable=False)
    checkout_date=sa.Column(types.DateTime, nullable=False)
    user_id = sa.Column(BIGINT, sa.ForeignKey("ums_users.id"), nullable=False)
    order_status=sa.Column(sa.Enum(MarketPlaceOrderStatus), nullable=False)
    order_amount = sa.Column(BIGINT)
    order_details = relationship("OrderDetails", back_populates="order", lazy="dynamic")

class OrderDetails(Model):
    __tablename__="order_details"
    id = sa.Column(BIGINT, primary_key=True, autoincrement=True)
    order_id=sa.Column(BIGINT, ForeignKey("marketplace_orders.id"), nullable=False)
    prod_id = sa.Column(BIGINT, ForeignKey("products.id"), nullable=False)
    price_id = sa.Column(BIGINT, ForeignKey("prices.id"), nullable=False)
    quantity=sa.Column(sa.Integer, nullable=False)
    amount=sa.Column(sa.Integer, nullable=False)
    order = relationship("Order", back_populates="order_details")

API调用参数 #

{
  "user_id": 1,
  "order_detail": [ { "prod_id": 1, "price_id": 3, "amount": 1200 }, { "prod_id": 1, "price_id": 3, "amount": 12233 } ]
}

问题原因分析 #

1. bulk_insert_mappings用法错误 #

bulk_insert_mappings的正确参数是：第一个参数为SQLAlchemy模型类（如OrderDetails），第二个参数为字典列表（而非模型实例列表）。原代码传入了Order实例和OrderDetails实例列表，完全不符合方法要求，因此触发类型错误。

2. 子表实例重复引用 #

循环中仅创建了一个OrderDetails对象od，每次循环仅修改其属性并添加到列表，导致列表中所有元素都是同一对象的引用。SQLAlchemy处理时只会保留最后一次修改的属性值，最终仅插入一条子表数据。

修正方案及代码 #

核心修改点 #

• 循环内每次创建新的OrderDetails实例，避免引用重复
• 使用with db.begin()上下文管理器自动处理事务提交/回滚
• 利用SQLAlchemy关系映射，添加父对象时自动插入关联子对象，无需手动操作子表
• 修正日期字段类型匹配问题（原代码用date()，数据表定义为DateTime）

修正后的业务代码 #

from sqlalchemy.exc import SQLAlchemyError
import random
import string
from datetime import datetime
from pydantic import BaseModel, List
from sqlalchemy.orm import Session
# 需提前导入Order、OrderDetails模型及相关依赖

class OrderDetailModel(BaseModel):
    prod_id: int
    price_id: int
    amount: int

class OrderModel(BaseModel):
    user_id: int
    order_detail: List[OrderDetailModel] = []

    def create_order(self, db: Session):
        try:
            with db.begin():
                order_details = []
                # 循环创建新的OrderDetails实例
                for item in self.order_detail:
                    od = OrderDetails(
                        prod_id=item.prod_id,
                        price_id=item.price_id,
                        quantity=1,
                        amount=item.amount
                    )
                    order_details.append(od)

                order = Order(
                    unique_order_id=''.join(random.choices(string.ascii_lowercase, k=10)),
                    order_date=datetime.utcnow(),
                    checkout_date=datetime.utcnow(),
                    user_id=self.user_id,
                    order_status="FINISHED",
                    order_amount=100,
                    order_details=order_details
                )
                db.add(order)  # 自动插入父表及关联子表
            return "success"
        except SQLAlchemyError as e:
            db.rollback()
            # 可添加错误日志或返回错误信息
            raise e

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