feat: Batch delete and move applications

[shaohuzhang1]

feat: Batch delete and move applications

[f2c-ci-robot]

Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our release note process to remove it.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[f2c-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[shaohuzhang1]

There are a few optimizations and corrections that can be made to the provided code. Here's a summary of the changes:

Corrections

1. Import Statement: In get_application_operation_object, replace:

   from application.objects.object import QuerySet

   with:

   from django.db.models.base import ModelBase

2. Batch Move Parameter Definition: Remove duplicate parameter definitions in put method.

3. Logging Functionality: Ensure that the logging function (get_operation_object) returns the correct structure based on the input parameters.

4. Function Name Consistency: Rename functions where necessary to make them more descriptive.

Here's the updated version of the relevant parts:

def get_application_operation_object_batch(application_id_list):
    # Assuming query_set is imported correctly here
    application_queryset = QuerySet(Application).filter(id__in=application_id_list)
    if application queryset.exists():
        return {
            "name": f"([{','.join(app.name for app in application_queryset)})]",
            'application_list': [{'name': app.name} for app in application_queryset]
        }
    return {}

# ... other unchanged methods ...

class BatchDelete(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['PUT'],
        description="Batch delete applications",
        summary="Batch delete applications",
        operation_id="Batch delete applications",
        parameters=ApplicationBatchOperateAPI.get_parameters(),
        request=ApplicationBatchOperateAPI.get_request(),
        responses=result.DefaultResultSerializer,
        tags=[_("Application")]
    )
    @has_permissions(
        PermissionConstants.APPLICATION_BATCH_DELETE.get_workspace_permission(), 
        RoleConstants.USER.get_workspace_role(),
        RoleConstants.WORKSPACE_MANAGE.get_workspace_role()
    )
    def put(self, request: Request, workspace_id: str):
        id_list = request.data.get('id_list', [])
        
        # Simplify permission check
        permitted_ids = list(filter(lambda x: x.is_valid(workspace_id), id_list))

        @log(menu='Application', operate='Batch delete applications')
        def inner(view, r, **kwargs):
            try:
                serializer = ApplicationBatchOperateSerializer(data={"workspace_id": workspace_id, "user_id": request.user.id})
                response_data = serializer.batch_delete({"id_list": permitted_ids})
                return result.success(response_data)
            except Exception as e:
                return result.fail(str(e))
            
        return inner(self, request)

class BatchMove(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['PUT'],
        description="Batch move applications",
        summary="Batch move applications",
        operation_id="Batch move applications",
        parameters=ApplicationBatchOperateAPI.get_parameters(),
        request=ApplicationBatchOperateAPI.get_move_request(),
        responses=result.DefaultResultSerializer,
        tags=[_("Application")]
    )
    @has_permissions(
        PermissionConstants.APPLICATION_BATCH_MOVE.get_workspace_permission(), 
        RoleConstants.USER.get_workspace_role(),
        RoleConstants.WORKSPACE_MANAGE.get_workspace_role()
    )
    def put(self, request: Request, workspace_id: str):
        id_list = request.data.get('id_list', [])
        
        # Simplify permission check
        permitted_ids = list(filter(lambda x: x.is_valid(workspace_id), id_list))

        folder_id = request.data.get('folder_id')

        @log(menu='Application', operate='Batch move applications')
        def inner(view, r, **kwargs):
            try:
                serializer = ApplicationBatchOperateSerializer(data={"workspace_id": workspace_id, "user_id": request.user.id})
                response_data = serializer.batch_move({"id_list": permitted_ids, "folder_id": folder_id})
                return result.success(response_data)
            except Exception as e:
                return result.fail(str(e))
            
        return inner(self, request)

Additional Suggestions

• Type Hinting and Documentation: Add type hints where appropriate and ensure comprehensive documentation.

By making these corrections, the code should be more robust, efficient, and easier to maintain.

[shaohuzhang1]

The provided code snippet contains several potential issues that need to be addressed:

1. Duplicated Code: The play_demo_text method exists in both ApplicationModelViewSet and ApplicationViewSet. This redundancy can lead to maintenance problems and unnecessary function calls.

2. Transaction Management: There is no transaction management applied on the delete methods (batch_delete). A transaction should be used around database operations to ensure atomicity and consistency.

3. Code Duplication: Similar logic exists between the batch_delete and batch_move methods. This duplication can be refactored into a single method if possible for better maintainability.

4. Error Handling: In some scenarios, exceptions are raised directly without handling them properly, which could lead to unanticipated behavior during production use.

5. Validation Logic: The validation logic in batch_delete and batch_move is complex and repetitive. Consider modularizing this logic further using mixins or custom validator classes.

Here's an optimized version of the code with suggested improvements:

from rest_framework import viewsets, serializers, status
from django.db import transaction
from models_provider.models import Model
from knowledge.models import (
    Application,
    ApplicationVersion,
    ResourceMapping,
    TriggerTask,
    Trigger
)
from knowledge.serializers.common import BatchSerializer, BatchMoveSerializer
from knowledge.serializers.knowledge import ApplicationSerializer
from maxkb.conf import PROJECT_DIR

class ApplicationModelViewSet(viewsets.ModelViewSet):
    queryset = Application.objects.all()
    serializer_class = ApplicationSerializer

    def play_demo_text(self, instance):
        tts_model_id = instance.pop('tts_model_id')
        model = get_model_instance_by_model_workspace_id(tts_model_id, self.data.get('workspace_id'), **instance)
        return model.text_to_speech(instance['text'])

    class ApplicationBatchOperateSerializer(serializers.Serializer):
        workspace_id = serializers.CharField(required=True)

        def save_batch(self, action, data: List[Dict], **kwargs) -> bool:
            with transaction.atomic():
                if action == 'DELETE':
                    result = self.batch_delete(data, **kwargs)
                elif action == 'MOVE':
                    result = self.batch_move(data, **kwargs)
                else:
                    raise NotImplementedError(f"Action '{action}' not supported.")
                return result

        @transaction.atomic
        def batch_delete(self, ids: List[int], with_valid=True) -> bool:
            # Validate input before processing
            instance = {'id_list': ids}
            if with_valid:
                BatchSerializer(data=instance, many=True).is_valid(raise_exception=True)
                self.is_valid(raise_exception=True)
            
            try:
                QuerySet(ApplicationVersion).filter(application_id__in=ids).delete()
                QuerySet(ResourceMapping).filter(
                    Q(target_id__in=ids) | Q(source_id__in=ids)
                ).delete()

                QuerySet(Application).filter(id__in=ids).delete()

                trigger_ids = list(QuerySet(TriggerTask).filter(
                    source_type="APPLICATION", source_id__in=ids
                ).values('trigger_id').distinct())

                QuerySet(TriggerTask).filter(source_type="APPLICATION", source_id__in=ids).delete()

                for trigger_id in trigger_ids:
                    trigger = Trigger.objects.filter(id=trigger_id['trigger_id'}).first()
                    if trigger and trigger.is_active:
                        deploy(TriggerModelSerializer(trigger).data, **{})

                return True
            except Exception as e:
                maxkb_logger.error(f"Failed to perform batch {action}", exc_info=e)
                return False

        @transaction.atomic
        def batch_move(self, new_folder_id: int, ids: List[int], **kwargs) -> bool:
            # Validate input before processing
            instance = {'id_list': ids, 'folder_id': new_folder_id}
            if kwargs.get('with_valid', True):
                BatchMoveSerializer(data=instance, many=True).is_valid(raise_exception=True)
                self.is_valid(raise_exception=True)

            try:
                QuerySet(Application).filter(id__in=ids).update(folder_id=new_folder_id)
                return True
            except Exception as e:
                maxkb_logger.error(f"Failed to move batch applications", exc_info=e)
                return False

This revised approach improves organization and reduces redundancy by encapsulating the business logic in a reusable save_batch method. Additionally, appropriate error logging is added within each operation to help diagnose failures and monitor application performance.

[shaohuzhang1]

There are no apparent irregularities, but here are a few minor optimizations and improvements:

1. Consistent Formatting: The file is well-formatted with consistent indentation.
2. Clear Documentation: Each enumeration value should have comments explaining its usage clearly. The current implementation does not include explanations for each operation.
3. Code Reusability: Combine repeated permission definitions where possible. For example, you can create a base PermissionBase class that includes common fields and then subclass it for specific permissions.

Here's an optimized version of the code including brief descriptions:

class Operation(Enum):
    READ = "READ"
    EDIT = "EDIT"
    CREATE = "CREATE"
    DELETE = "DELETE"
    BATCH_DELETE = "BATCH_DELETE"
    BATCH_MOVE = "BATCH_MOVE"


class Group(Enum):
    TOOL = "TOOL"
    KNOWLEDGE = "KNOWLEDGE"
    APPLICATION = "APPLICATION"



def get_role_group(role_name: str) -> RoleGroup:
    roles_dict = {
        RoleConstants.ADMIN.value: RoleGroup.ADMIN,
        RoleConstants.USER.value: RoleGroup.USER
    }
    return roles_dict.get(role_name.upper())


# Base Permission class
class PermissionBase(Enum):
    def __init__(self, group: Group, operate: Operation, role_list: list):
        self.group = group
        self.operate = operate
        self.role_list = role_list

class Permission(PermissionBase):
    def __new__(cls, *args, **kwargs):
        obj = super().__new__(cls)
        # Add additional logic if needed, e.g., default resource_permission_group_list
        kwargs.update({"resource_permission_group_list": []})
        obj.__dict__.update(kwargs)
        return obj


class ToolPermissionsEnum:
    @staticmethod
    def batch_permissions():
        return [
            Permission(group=Group.TOOL, operate=Operation.BATCH_DELETE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")]),
            Permission(group=Group.TOOL, operate=Operation.BATCH_MOVE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")])
        ]

class KnowledgePermissionsEnum:
    @staticmethod
    def batch_permissions():
        return [
            Permission(group=Group.KNOWLEDGE, operate=Operation.BATCH_DELETE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")]),
            Permission(group=Group.KNOWLEDGE, operate=Operation.BATCH_MOVE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")])
        ]


class ApplicationPermissionsEnum:
    @staticmethod
    def batch_permissions():
        return [
            Permission(group=Group.APPLICATION, operate=Operation.BATCH_DELETE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")]),
            Permission(group=Group.APPLICATION, operate=Operation.BATCH_MOVE,
                       role_list=[get_role_group("ADMIN"), get_role_group("USER")])
        ]

This refactoring improves readability and reusability by creating separate enums for different types (e.g., tool permissions), which can be iterated upon easily using static methods.