mirror of
https://github.com/1Panel-dev/MaxKB.git
synced 2025-12-26 10:12:51 +00:00
270 lines
14 KiB
Python
270 lines
14 KiB
Python
# coding=utf-8
|
||
"""
|
||
@project: MaxKB
|
||
@Author：虎虎
|
||
@file： user.py
|
||
@date：2025/4/14 19:25
|
||
@desc:
|
||
"""
|
||
from django.db.models import QuerySet
|
||
from drf_spectacular.utils import extend_schema
|
||
from rest_framework.views import APIView
|
||
from django.utils.translation import gettext_lazy as _
|
||
from rest_framework.request import Request
|
||
|
||
from common.auth import TokenAuth
|
||
from common.auth.authentication import has_permissions
|
||
from common.constants.permission_constants import PermissionConstants, RoleConstants
|
||
from common.log.log import log
|
||
from common.result import result
|
||
from common.utils.common import query_params_to_single_dict
|
||
from models_provider.api.model import ModelCreateAPI, GetModelApi, ModelEditApi, ModelListResponse, DefaultModelResponse
|
||
from models_provider.api.provide import ProvideApi
|
||
from models_provider.models import Model
|
||
from models_provider.serializers.model_serializer import ModelSerializer, \
|
||
WorkspaceSharedModelSerializer
|
||
from system_manage.views import encryption_str
|
||
|
||
|
||
def encryption_credential(credential):
|
||
if isinstance(credential, dict):
|
||
return {key: encryption_str(credential.get(key)) for key in credential}
|
||
return credential
|
||
|
||
|
||
def get_edit_model_details(request):
|
||
path = request.path
|
||
body = request.data
|
||
query = request.query_params
|
||
credential = body.get('credential', {})
|
||
credential_encryption_ed = encryption_credential(credential)
|
||
return {
|
||
'path': path,
|
||
'body': {**body, 'credential': credential_encryption_ed},
|
||
'query': query
|
||
}
|
||
|
||
|
||
def get_model_operation_object(model_id):
|
||
model_model = QuerySet(model=Model).filter(id=model_id).first()
|
||
if model_model is not None:
|
||
return {
|
||
"name": model_model.name
|
||
}
|
||
return {}
|
||
|
||
|
||
class ModelSetting(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['POST'],
|
||
summary=_("Create model"),
|
||
description=_("Create model"),
|
||
operation_id=_("Create model"), # type: ignore
|
||
tags=[_("Model")], # type: ignore
|
||
parameters=ModelCreateAPI.get_parameters(),
|
||
request=ModelCreateAPI.get_request(),
|
||
responses=ModelCreateAPI.get_response())
|
||
@has_permissions(PermissionConstants.MODEL_CREATE.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
@log(menu='model', operate='Create model',
|
||
get_operation_object=lambda r, k: {'name': r.date.get('name')},
|
||
get_details=get_edit_model_details,
|
||
)
|
||
def post(self, request: Request, workspace_id: str):
|
||
return result.success(
|
||
ModelSerializer.Create(data={**request.data, 'user_id': request.user.id}).insert(workspace_id,
|
||
with_valid=True))
|
||
|
||
# @extend_schema(methods=['PUT'],
|
||
# summary=_('Update model'),
|
||
# operation_id=_('Update model'), # type: ignore
|
||
# request=ModelEditApi.get_request(),
|
||
# responses=ModelCreateApi.get_response(),
|
||
# tags=[_('Model')]) # type: ignore
|
||
# @has_permissions(PermissionConstants.MODEL_CREATE)
|
||
# def put(self, request: Request):
|
||
# return result.success(
|
||
# ModelSerializer.Create(data={**request.data, 'user_id': str(request.user.id)}).insert(request.user.id,
|
||
# with_valid=True))
|
||
|
||
@extend_schema(methods=['GET'],
|
||
summary=_('Query model list'),
|
||
description=_('Query model list'),
|
||
operation_id=_('Query model list'), # type: ignore
|
||
parameters=ModelListResponse.get_parameters(),
|
||
responses=ModelListResponse.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def get(self, request: Request, workspace_id: str):
|
||
return result.success(
|
||
ModelSerializer.Query(
|
||
data={**query_params_to_single_dict(request.query_params)}).list(workspace_id=workspace_id,
|
||
with_valid=True))
|
||
|
||
class Operate(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['PUT'],
|
||
summary=_('Update model'),
|
||
description=_('Update model'),
|
||
operation_id=_('Update model'), # type: ignore
|
||
request=ModelEditApi.get_request(),
|
||
parameters=GetModelApi.get_parameters(),
|
||
responses=ModelEditApi.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_EDIT.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
@log(menu='model', operate='Update model',
|
||
get_operation_object=lambda r, k: get_model_operation_object(k.get('model_id')),
|
||
get_details=get_edit_model_details,
|
||
)
|
||
def put(self, request: Request, workspace_id, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.Operate(
|
||
data={'id': model_id, 'user_id': request.user.id, 'workspace_id': workspace_id}).edit(request.data,
|
||
str(request.user.id)))
|
||
|
||
@extend_schema(methods=['DELETE'],
|
||
summary=_('Delete model'),
|
||
description=_('Delete model'),
|
||
operation_id=_('Delete model'), # type: ignore
|
||
parameters=GetModelApi.get_parameters(),
|
||
responses=DefaultModelResponse.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_DELETE.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
@log(menu='model', operate='Delete model',
|
||
get_operation_object=lambda r, k: get_model_operation_object(k.get('model_id')),
|
||
)
|
||
def delete(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.Operate(
|
||
data={'id': model_id, 'user_id': request.user.id, 'workspace_id': workspace_id}).delete())
|
||
|
||
@extend_schema(methods=['GET'],
|
||
summary=_('Query model details'),
|
||
description=_('Query model details'),
|
||
operation_id=_('Query model details'), # type: ignore
|
||
parameters=GetModelApi.get_parameters(),
|
||
responses=GetModelApi.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def get(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.Operate(
|
||
data={'id': model_id, 'user_id': request.user.id, 'workspace_id': workspace_id}).one(
|
||
with_valid=True))
|
||
|
||
class ModelParamsForm(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['GET'],
|
||
summary=_('Get model parameter form'),
|
||
description=_('Get model parameter form'),
|
||
operation_id=_('Get model parameter form'), # type: ignore
|
||
parameters=GetModelApi.get_parameters(),
|
||
responses=ProvideApi.ModelParamsForm.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def get(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.ModelParams(data={'id': model_id}).get_model_params())
|
||
|
||
@extend_schema(methods=['PUT'],
|
||
summary=_('Save model parameter form'),
|
||
description=_('Save model parameter form'),
|
||
operation_id=_('Save model parameter form'), # type: ignore
|
||
parameters=GetModelApi.get_parameters(),
|
||
request=GetModelApi.get_request(),
|
||
responses=ProvideApi.ModelParamsForm.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
@log(menu='model', operate='Save model parameter form',
|
||
get_operation_object=lambda r, k: get_model_operation_object(k.get('model_id')),
|
||
)
|
||
def put(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.ModelParams(data={'id': model_id}).save_model_params_form(request.data))
|
||
|
||
class ModelMeta(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['GET'],
|
||
summary=_(
|
||
'Query model meta information, this interface does not carry authentication information'),
|
||
description=_(
|
||
'Query model meta information, this interface does not carry authentication information'),
|
||
operation_id=_(
|
||
'Query model meta information, this interface does not carry authentication information'),
|
||
parameters=GetModelApi.get_parameters(),
|
||
responses=GetModelApi.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def get(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.Operate(data={'id': model_id, 'workspace_id': workspace_id}).one_meta(with_valid=True))
|
||
|
||
class PauseDownload(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['PUT'],
|
||
summary=_('Pause model download'),
|
||
description=_('Pause model download'),
|
||
operation_id=_('Pause model download'), # type: ignore
|
||
parameters=GetModelApi.get_parameters(),
|
||
request=GetModelApi.get_request(),
|
||
responses=DefaultModelResponse.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_CREATE.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def put(self, request: Request, workspace_id: str, model_id: str):
|
||
return result.success(
|
||
ModelSerializer.Operate(data={'id': model_id, 'workspace_id': workspace_id}).pause_download())
|
||
|
||
|
||
class WorkspaceSharedModelSetting(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(
|
||
methods=['Get'],
|
||
summary=_('Get Share model by workspace id'),
|
||
description=_('Get Share model by workspace id'),
|
||
operation_id=_('Get Share model by workspace id'), # type: ignore
|
||
parameters=ModelListResponse.get_parameters(),
|
||
responses=DefaultModelResponse.get_response(),
|
||
tags=[_('Shared Model')]
|
||
) # type: ignore
|
||
@has_permissions(
|
||
PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
|
||
RoleConstants.USER.get_workspace_role(),
|
||
)
|
||
def get(self, request: Request, workspace_id: str):
|
||
return result.success(
|
||
WorkspaceSharedModelSerializer(data={'workspace_id': workspace_id}).get_share_model_list())
|
||
|
||
|
||
class ModelList(APIView):
|
||
authentication_classes = [TokenAuth]
|
||
|
||
@extend_schema(methods=['GET'],
|
||
summary=_('Query all model list'),
|
||
description=_('Query all model list'),
|
||
operation_id=_('Query all model list'), # type: ignore
|
||
parameters=ModelListResponse.get_parameters(),
|
||
responses=ModelListResponse.get_response(),
|
||
tags=[_('Model')]) # type: ignore
|
||
@has_permissions(PermissionConstants.MODEL_READ.get_workspace_permission(),
|
||
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), RoleConstants.USER.get_workspace_role())
|
||
def get(self, request: Request, workspace_id: str):
|
||
return result.success(
|
||
ModelSerializer.Query(
|
||
data={**query_params_to_single_dict(request.query_params)}).model_list(workspace_id=workspace_id,
|
||
with_valid=True))
|