基于基础DocType在Frappe中向其他DocType同步数据优化需求

需求概述 #

• 保存Employee文档时，自动同步其关联Job Role的课程到Course Assignment、技能到Employee Skill
• 仅同步未已存在的记录（按员工+课程/员工+技能匹配）
• 解决首次保存Employee时提示"DocType不存在"的问题
• 优化原有循环逻辑，提升代码性能

原代码性能瓶颈 #

1. 循环内多次调用frappe.get_doc获取Job Role数据，重复查询数据库
2. 每条记录都调用frappe.db.exists，产生大量单次数据库请求
3. 循环内重复查询Course Template的start_date，冗余耗时

优化后的完整代码 #

import frappe
from frappe import _
from frappe.model.document import Document


class Employee(Document):
    def before_save(self):
        # 自动拼接员工全名
        if not self.full_name:
            name_parts = []
            if self.first_name:
                name_parts.append(self.first_name)
            if self.middle_name:
                name_parts.append(self.middle_name)
            if self.last_name:
                name_parts.append(self.last_name)
            self.full_name = ' '.join(name_parts).strip()
        
        # 判断Job Roles是否变更，或首次保存时触发同步
        job_roles_changed = False
        if self._doc_before_save:
            # 用集合对比前后Job Roles，高效判断变更
            old_job_roles = {jr.job_role for jr in self._doc_before_save.job_roles}
            new_job_roles = {jr.job_role for jr in self.job_roles}
            job_roles_changed = old_job_roles != new_job_roles
        
        if job_roles_changed or not self._doc_before_save:
            self.trigger_job_roles()

    def validate(self):
        # 校验工作日期逻辑
        if self.work_start_date and self.work_end_date:
            if self.work_start_date >= self.work_end_date:
                frappe.throw(_('工作结束日期必须晚于工作开始日期'))

    def trigger_job_roles(self):
        # 异步执行同步任务，避免阻塞保存操作，指定short队列优先处理
        frappe.enqueue(
            "medad_tms.trainee_management.doctype.employee.employee.assign_employee",
            doc=self,
            queue="short"
        )


def assign_employee(doc):
    try:
        # 收集所有关联的Job Role名称
        job_role_names = [jr.job_role for jr in doc.job_roles]
        if not job_role_names:
            return

        # 批量获取所有Job Role的课程和技能数据，避免循环内重复查询
        job_roles_data = frappe.db.get_list(
            "Job Role",
            filters={"name": ["in", job_role_names]},
            fields=["name", "required_courses", "required_skills"]
        )

        # 批量查询该员工已有的课程分配和技能记录，替代循环内的exists查询
        existing_course_assignments = frappe.db.get_list(
            "Course Assignment",
            filters={"trainee": doc.related_user},
            pluck="course"
        )
        existing_employee_skills = frappe.db.get_list(
            "Employee Skill",
            filters={"employee": doc.name},
            pluck="skill"
        )

        # 批量获取所有课程模板的start_date，避免循环内重复查询
        required_courses = []
        for jr in job_roles_data:
            required_courses.extend([item.course for item in jr.required_courses])
        
        course_start_dates = {}
        if required_courses:
            course_templates = frappe.db.get_list(
                "Course Template",
                filters={"name": ["in", required_courses]},
                fields=["name", "start_date"]
            )
            course_start_dates = {ct.name: ct.start_date for ct in course_templates}

        # 处理课程分配：仅创建未存在的记录
        for jr in job_roles_data:
            for course_item in jr.required_courses:
                course_name = course_item.course
                if course_name not in existing_course_assignments:
                    course_doc = frappe.new_doc("Course Assignment")
                    course_doc.trainee = doc.related_user
                    course_doc.course = course_name
                    course_doc.source = "Job Role"
                    course_doc.due_date = course_start_dates.get(course_name)
                    course_doc.insert(ignore_permissions=True)
                    existing_course_assignments.append(course_name)

        # 处理技能分配：仅创建未存在的记录
        for jr in job_roles_data:
            for skill_item in jr.required_skills:
                skill_name = skill_item.skill
                if skill_name not in existing_employee_skills:
                    skill_doc = frappe.new_doc("Employee Skill")
                    skill_doc.employee = doc.name
                    skill_doc.skill = skill_name
                    skill_doc.skill_type = "Training Programs"
                    skill_doc.proficiency_scale_level = 1
                    skill_doc.required_scale_level = 5
                    skill_doc.insert(ignore_permissions=True)
                    existing_employee_skills.append(skill_name)

        frappe.db.commit()
        frappe.publish_realtime(
            "assign_employee",
            {"progress": 1, "total": 3,
                "message": "已为员工分配课程和技能"},
            user=frappe.session.user,
            after_commit=True,
        )
    except Exception as e:
        frappe.db.rollback()
        frappe.log_error(frappe.get_traceback(), "Employee 同步失败")
        frappe.throw(_("分配课程和技能时出错：{0}").format(str(e)))

优化说明 #

1. 批量查询减少数据库请求：

   • 一次性获取所有关联Job Role的课程和技能数据，替代循环内的frappe.get_doc
   • 用pluck批量查询员工已有的记录，替代循环内的frappe.db.exists，减少N次数据库请求为2次
   • 批量获取课程模板的start_date，避免循环内重复查询

2. 逻辑优化：

   • 用集合对比Job Roles变更，比直接对比列表更高效准确
   • 异步任务指定short队列，优先处理同步操作
   • 插入记录后更新已存在列表，避免重复判断
   • 增加空值判断，无关联Job Role时直接返回，减少无效操作

3. 首次保存问题处理：

   • 保留原before_save中首次保存（_doc_before_save为None）时触发同步的逻辑
   • 异步执行任务，避免首次保存时DocType未完全初始化导致的错误

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