Method insert

apps/trigger/serializers/trigger.py:351–399

(self, instance, with_valid=True)

    @transaction.atomic
    def insert(self, instance, with_valid=True):
        from trigger.handler.simple_tools import deploy

        if with_valid:
            self.is_valid(raise_exception=True)
        serializer = TriggerCreateRequest(data=instance)
        serializer.is_valid(raise_exception=True)
        valid_data = serializer.validated_data

        trigger_id = valid_data.get('id') if valid_data.get('id') else uuid.uuid7()

        trigger_model = Trigger(
            id=trigger_id,
            name=valid_data.get('name'),
            workspace_id=self.data.get('workspace_id'),
            desc=valid_data.get('desc') or '',
            trigger_type=valid_data.get('trigger_type'),
            trigger_setting=valid_data.get('trigger_setting'),
            meta=valid_data.get('meta', {}),
            is_active=valid_data.get('is_active') or False,
            user_id=self.data.get('user_id'),
        )
        trigger_model.save()

        trigger_tasks = valid_data.get('trigger_task')
        if trigger_tasks:

            is_active_map = self.batch_get_source_active_status(trigger_tasks)

            trigger_task_models = [
                TriggerTask(
                    id=uuid.uuid7(),
                    trigger_id=trigger_id,
                    source_type=task_data.get('source_type'),
                    source_id=task_data.get('source_id'),
                    is_active=is_active_map.get((task_data.get('source_type'), task_data.get('source_id'))) or False,
                    parameter=task_data.get('parameter', {}),
                    meta=task_data.get('meta', {})
                )
                for task_data in trigger_tasks
            ]

            TriggerTask.objects.bulk_create(trigger_task_models)
        else:
            raise AppApiException(500, _('Trigger task can not be empty'))

        if trigger_model.is_active:
            deploy(TriggerModelSerializer(trigger_model).data, **{})
        return TriggerResponse(trigger_model).data

    @staticmethod
    def batch_get_source_active_status(trigger_tasks: list) -> Dict[tuple, bool]: