MaxKB/apps/application/views/application_api_key.py at f5517ebc1280f2b2f431f4e27a0ac987979fb889 · 1Panel-dev/MaxKB · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
from django.db.models import QuerySet
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.request import Request
from rest_framework.views import APIView

from application.api.application_api_key import ApplicationKeyAPI
from application.models import Application
from application.serializers.application_api_key import ApplicationKeySerializer
from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import PermissionConstants, RoleConstants, ViewPermission, CompareConstants
from common.log.log import log
from common.result import result, DefaultResultSerializer


def get_application_operation_object(application_id):
    application_model = QuerySet(model=Application).filter(id=application_id).first()
    if application_model is not None:
        return {
            "name": application_model.name
        }
    return {}


class ApplicationKey(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['POST'],
        description=_('Create application ApiKey'),
        summary=_('Create application ApiKey'),
        operation_id=_('Create application ApiKey'),  # type: ignore
        parameters=ApplicationKeyAPI.get_parameters(),
        request=None,
        responses=ApplicationKeyAPI.get_response(),
        tags=[_('Application Api Key')]  # type: ignore
    )
    @log(menu='Application', operate="Add ApiKey",
         get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')),
         )
    @has_permissions(PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_application_permission(),
                     PermissionConstants.APPLICATION_READ.get_workspace_permission_workspace_manage_role(),
                     ViewPermission([RoleConstants.USER.get_workspace_role()],
                                    [PermissionConstants.APPLICATION.get_workspace_application_permission()],
                                    CompareConstants.AND),
                     RoleConstants.WORKSPACE_MANAGE.get_workspace_role()
                     )
    def post(self, request: Request, workspace_id: str, application_id: str):
        return result.success(ApplicationKeySerializer(
            data={'application_id': application_id,
                  'workspace_id': workspace_id,
                  'user_id': request.user.id}).generate())

    class Page(APIView):
        authentication_classes = [TokenAuth]

        @extend_schema(
            methods=['GET'],
            description=_('GET application ApiKey List'),
            summary=_('Create application ApiKey List'),
            operation_id=_('Create application ApiKey List'),  # type: ignore
            parameters=ApplicationKeyAPI.get_parameters(),
            responses=ApplicationKeyAPI.List.get_response(),
            tags=[_('Application Api Key')]  # type: ignore
        )
        @has_permissions(PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_application_permission(),
                         PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_permission_workspace_manage_role(),
                         ViewPermission([RoleConstants.USER.get_workspace_role()],
                                        [PermissionConstants.APPLICATION.get_workspace_application_permission()],
                                        CompareConstants.AND),
                         RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
        def get(self, request: Request, workspace_id: str, application_id: str, current_page: int, page_size: int):
            return result.success(ApplicationKeySerializer(
                data={'application_id': application_id,
                      'workspace_id': workspace_id,
                      'order_by': request.query_params.get('order_by')}).page(current_page, page_size))

    class Operate(APIView):
        authentication_classes = [TokenAuth]

        @extend_schema(
            methods=['PUT'],
            description=_('Modify application API_KEY'),
            summary=_('Modify application API_KEY'),
            operation_id=_('Modify application API_KEY'),  # type: ignore
            parameters=ApplicationKeyAPI.Operate.get_parameters(),
            request=ApplicationKeyAPI.Operate.get_request(),
            responses=DefaultResultSerializer,
            tags=[_('Application Api Key')]  # type: ignore
        )
        @has_permissions(PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_application_permission(),
                         PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_permission_workspace_manage_role(),
                         ViewPermission([RoleConstants.USER.get_workspace_role()],
                                        [PermissionConstants.APPLICATION.get_workspace_application_permission()],
                                        CompareConstants.AND),
                         RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
        @log(menu='Application', operate="Modify application API_KEY",
             get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')),
             )
        def put(self, request: Request, workspace_id: str, application_id: str, api_key_id: str):
            return result.success(
                ApplicationKeySerializer.Operate(
                    data={'workspace_id': workspace_id, 'application_id': application_id,
                          'api_key_id': api_key_id}).edit(
                    request.data))

        @extend_schema(
            methods=['DELETE'],
            description=_('Delete Application API_KEY'),
            summary=_('Delete Application API_KEY'),
            operation_id=_('Delete Application API_KEY'),  # type: ignore
            parameters=ApplicationKeyAPI.Operate.get_parameters(),
            request=ApplicationKeyAPI.Operate.get_request(),
            responses=DefaultResultSerializer,
            tags=[_('Application Api Key')]  # type: ignore
        )
        @has_permissions(PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_application_permission(),
                         PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_permission_workspace_manage_role(),
                         ViewPermission([RoleConstants.USER.get_workspace_role()],
                                        [PermissionConstants.APPLICATION.get_workspace_application_permission()],
                                        CompareConstants.AND),
                         RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
        @log(menu='Application', operate="Delete application API_KEY",
             get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')),
             )
        def delete(self, request: Request, workspace_id: str, application_id: str, api_key_id: str):
            return result.success(
                ApplicationKeySerializer.Operate(
                    data={'workspace_id': workspace_id, 'application_id': application_id,
                          'api_key_id': api_key_id}).delete())