def METHOD_NAME(self, **kwargs):
    task_name = kwargs["task_name"]
    plan_start_time = kwargs["plan_start_time"]
    creator = kwargs["creator"]
    project_id = kwargs["project_id"]
    template_id = kwargs["template_id"]
    task_params = kwargs["task_params"]
    template_name = kwargs["template_name"]
    notify_type = kwargs.get("notify_type", "[]")
    notify_receivers = kwargs.get("notify_receivers", "{}")
    optional_keys = ["editor", "edit_time", "create_time"]
    # 过滤 optional_keys 中不存在于 kwargs 的属性
    extra_data = {key: kwargs[key] for key in filter(lambda x: x in kwargs, optional_keys)}
    with transaction.atomic():
        clocked, _ = DjangoCeleryBeatClockedSchedule.objects.get_or_create(clocked_time=plan_start_time)
        task = ClockedTask.objects.create(
            project_id=project_id,
            template_id=template_id,
            task_name=task_name,
            template_name=template_name,
            creator=creator,
            plan_start_time=plan_start_time,
            task_params=task_params,
            notify_type=notify_type,
            notify_receivers=notify_receivers,
            **extra_data,
        )
        clocked_task_kwargs = {"clocked_task_id": task.id}
        clocked_task = DjangoCeleryBeatPeriodicTask.objects.create(
            clocked=clocked,
            name=task_name + uniqid(),
            task="gcloud.clocked_task.tasks.clocked_task_start",
            one_off=True,
            kwargs=json.dumps(clocked_task_kwargs),
        )
        task.clocked_task_id = clocked_task.id
        task.save()
    return task