MaxKB/apps/application/views/application_api_key.py
wxg0103 e6e3acdfea
Some checks are pending
sync2gitee / repo-sync (push) Waiting to run
refactor: log add workspace_id
2025-06-10 14:50:25 +08:00

106 lines
5.1 KiB
Python

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 ApplicationApiKey, 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
from common.log.log import log
from common.result import result, success, 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')),
workspace_id=lambda r, k: k.get('workspace_id'))
@has_permissions(PermissionConstants.APPLICATION_OVERVIEW_API_KEY.get_workspace_application_permission())
def post(self, request: Request, workspace_id: str, application_id: str):
return result.success(ApplicationKeySerializer(
data={'application_id': application_id, 'user_id': request.user.id,
'workspace_id': workspace_id}).generate())
@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())
def get(self, request: Request, workspace_id: str, application_id: str):
return result.success(ApplicationKeySerializer(
data={'application_id': application_id,
'workspace_id': workspace_id}).list())
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())
@log(menu='Application', operate="Modify application API_KEY",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')),
workspace_id=lambda r, k: k.get('workspace_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())
@log(menu='Application', operate="Delete application API_KEY",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')),
workspace_id=lambda r, k: k.get('workspace_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())