Odoo 12企业版：将ORM单条插入改为PostgreSQL批量插入优化性能

问题背景 #

使用Odoo 12企业版时，调用action_confirm_1函数创建mrp.production记录耗时超40秒，目标将耗时压缩至20秒内。当前代码通过循环单条调用create方法插入数据，同时需要递归设置子生产订单的plan_order_id字段，性能瓶颈在于单条ORM操作的重复开销。

优化思路 #

将单条ORM创建改为PostgreSQL批量SQL插入，减少数据库交互次数；同时通过PostgreSQL递归CTE查询批量更新子生产订单的plan_order_id，替代原Python递归循环+单条写入的低效逻辑。

优化后代码实现 #

@api.multi
def action_confirm_1(self):
    request_plan = self.filtered(lambda r: r.state == 'checked')
    if not request_plan:
        return False
    
    # 批量收集所有主生产订单的插入数据
    batch_vals = []
    plan_production_map = {}  # 记录plan.id对应的主生产订单ID列表
    
    for plan in request_plan:
        current_plan_vals = []
        for line in plan.plan_line_ids:
            vals = line._prepare_production_vals(picking_type_id=plan.picking_type_id)
            # 转换字段格式适配SQL插入
            if isinstance(vals.get('date_planned_start'), datetime.datetime):
                vals['date_planned_start'] = fields.Datetime.to_string(vals['date_planned_start'])
            if isinstance(vals.get('date_planned_finished'), datetime.datetime):
                vals['date_planned_finished'] = fields.Datetime.to_string(vals['date_planned_finished'])
            # 将False替换为SQL兼容的NULL
            for key, value in vals.items():
                if value is False:
                    vals[key] = None
            current_plan_vals.append(vals)
        batch_vals.extend(current_plan_vals)
        # 先暂存关联关系，后续用插入后的实际ID替换
        plan_production_map[plan.id] = []
    
    if batch_vals:
        # 获取插入字段列表（排除自增id）
        fields = list(batch_vals[0].keys())
        # 批量插入主生产订单并返回插入记录ID
        self.env.cr.execute(
            f"""
            INSERT INTO mrp.production ({', '.join(fields)})
            VALUES {', '.join(['%s'] * len(batch_vals))}
            RETURNING id, plan_order_id
            """,
            [tuple(vals[field] for field in fields) for vals in batch_vals]
        )
        inserted_productions = self.env.cr.fetchall()
        
        # 更新plan与主生产订单的关联映射
        for prod_id, plan_id in inserted_productions:
            plan_production_map[plan_id].append(prod_id)
        
        # 批量更新所有层级子生产订单的plan_order_id
        for plan_id, main_prod_ids in plan_production_map.items():
            if not main_prod_ids:
                continue
            self.env.cr.execute(
                """
                WITH RECURSIVE child_prods AS (
                    SELECT id FROM mrp.production WHERE id IN %s
                    UNION ALL
                    SELECT mp.id FROM mrp.production mp
                    JOIN child_prods cp ON mp.parent_id = cp.id
                )
                UPDATE mrp.production
                SET plan_order_id = %s
                WHERE id IN (SELECT id FROM child_prods)
                """,
                (tuple(main_prod_ids), plan_id)
            )
    
    # 更新计划状态
    request_plan.write({'state': 'confirm'})
    return True

关键优化点说明 #

• 批量SQL插入：一次性插入所有主生产订单，避免ORM循环调用带来的权限检查、信号触发、日志记录等重复开销
• 递归CTE查询子订单：利用PostgreSQL原生递归查询能力，一次性获取所有层级的子生产订单并批量更新，替代原Python递归循环+多次write操作
• 字段格式适配：统一转换Python数据类型为SQL兼容格式，避免插入报错

注意事项 #

1. 需确保_prepare_production_vals生成的字段与mrp.production表字段完全匹配，避免字段缺失或冗余
2. 若生产订单创建时依赖ORM钩子（如create信号、工作流触发），需评估批量SQL插入是否会跳过这些逻辑，必要时手动补充处理
3. 建议在测试环境验证数据正确性后再部署到生产环境

原问题相关代码参考 #

当前核心代码 #

@api.multi
def action_confirm_1(self):
    # start_time = time.monotonic()
    request_plan = self.filtered(lambda r: r.state == 'checked')
    if not request_plan:
        return False
    for plan in request_plan:
        Production = self.env['mrp.production']
        ProductionSudo = Production.sudo(self.env.user).with_context(force_company=plan.company_id.id)
        for line in plan.plan_line_ids:
            vals = line._prepare_production_vals(picking_type_id=plan.picking_type_id)
            production = ProductionSudo.create(vals)
            if production.child_production_ids:
                child_production = plan.get_child_production_ids(production)
                if child_production:
                    child_production.write({'plan_order_id': plan.id})
    request_plan.write({'state': 'confirm'})
    return True

vals变量示例 #

{'origin': False, 
'product_id': 47322, 
'product_qty': 1000.0, 
'product_uom_id': 8, 
'location_src_id': 54, 
'location_dest_id': 54, 'bom_id': 46430, 
'date_planned_start': datetime.datetime(2024, 7, 8, 1, 14, 2), 
'date_planned_finished': datetime.datetime(2024, 7, 8, 1, 14, 2), 
'procurement_group_id': False,
 'propagate': False, 
 'picking_type_id': 32, 
 'company_id': 4, 
 'user_id': 146, 
 'plan_order_id': 6515}

辅助函数代码 #

_prepare_production_vals

@api.multi
def _prepare_production_vals(self, picking_type_id):
    self.ensure_one()
    company_id = self.plan_id.company_id or self.env.user.company_id
    if picking_type_id.default_location_src_id:
        location_id = picking_type_id.default_location_src_id.id
    else:
        location_id = self._get_vals_location_id(
            company_id=company_id)
    if picking_type_id.default_location_dest_id:
        location_dest_id = picking_type_id.default_location_dest_id.id
    else:
        location_dest_id = self._get_vals_location_id(
            company_id=company_id)
    return {
        'origin': self.origin,
        'product_id': self.product_id.id,
        'product_qty': self.product_qty,
        'product_uom_id': self.product_uom_id.id,
        'location_src_id': location_id,
        'location_dest_id': location_dest_id,
        'bom_id': self.bom_id.id,
        'date_planned_start': self.schedule_date or fields.Datetime.now,
        'date_planned_finished': self.schedule_date or fields.Datetime.now,
        'procurement_group_id': False,
        'propagate': False,
        'picking_type_id': picking_type_id.id or self.with_context(
            company_id=company_id)._get_vals_picking_type(),
        'company_id': company_id.id,
        'user_id': self.plan_id.user_id and self.plan_id.user_id.id or \
            self.env.user.id,
        'plan_order_id': self.plan_id.id
    }

get_child_production_ids

def get_child_production_ids(self, production):
    child_production = self.env['mrp.production']
    queue = [(production, False)]  # (production, visited)
    while queue:
        current_production, visited = queue.pop(0)
        if not visited:
            child_production |= current_production
            queue.extend((child, False) for child in current_production.child_production_ids)
            queue.append((current_production, True))
    return child_production

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