15308555518/ MaxKB
1
0
Fork 0
mirror of https://github.com/1Panel-dev/MaxKB.git synced 2025-12-28 23:32:48 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: add import and export endpoints to ToolView for workspace tools

This commit is contained in:
CaptainB 2025-04-22 14:24:44 +08:00 committed by 刘瑞斌
parent 55705593a9
commit bbd7079166
7 changed files with 537 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/common/constants/permission_constants.py
View File

@ -139,6 +139,12 @@ class PermissionConstants(Enum):
RoleConstants.USER])
TOOL_DELETE = Permission(group=Group.TOOL, operate=Operate.DELETE, role_list=[RoleConstants.ADMIN,
RoleConstants.USER])
TOOL_DEBUG = Permission(group=Group.TOOL, operate=Operate.USE, role_list=[RoleConstants.ADMIN,
RoleConstants.USER])
TOOL_IMPORT = Permission(group=Group.TOOL, operate=Operate.USE, role_list=[RoleConstants.ADMIN,
RoleConstants.USER])
TOOL_EXPORT = Permission(group=Group.TOOL, operate=Operate.USE, role_list=[RoleConstants.ADMIN,
RoleConstants.USER])
def get_workspace_application_permission(self):
return lambda r, kwargs: Permission(group=self.value.group, operate=self.value.operate,

93
apps/common/utils/tool_code.py Normal file
View File

@ -0,0 +1,93 @@
# coding=utf-8
import os
import subprocess
import sys
import uuid
from textwrap import dedent
from diskcache import Cache
from maxkb.const import BASE_DIR
from maxkb.const import PROJECT_DIR
python_directory = sys.executable
class ToolExecutor:
def __init__(self, sandbox=False):
self.sandbox = sandbox
if sandbox:
self.sandbox_path = '/opt/maxkb/app/sandbox'
self.user = 'sandbox'
else:
self.sandbox_path = os.path.join(PROJECT_DIR, 'data', 'sandbox')
self.user = None
self._createdir()
if self.sandbox:
os.system(f"chown -R {self.user}:root {self.sandbox_path}")
def _createdir(self):
old_mask = os.umask(0o077)
try:
os.makedirs(self.sandbox_path, 0o700, exist_ok=True)
finally:
os.umask(old_mask)
def exec_code(self, code_str, keywords):
_id = str(uuid.uuid1())
success = '{"code":200,"msg":"成功","data":exec_result}'
err = '{"code":500,"msg":str(e),"data":None}'
path = r'' + self.sandbox_path + ''
_exec_code = f"""
try:
import os
env = dict(os.environ)
for key in list(env.keys()):
if key in os.environ and (key.startswith('MAXKB') or key.startswith('POSTGRES') or key.startswith('PG')):
del os.environ[key]
locals_v={'{}'}
keywords={keywords}
globals_v=globals()
exec({dedent(code_str)!a}, globals_v, locals_v)
f_name, f = locals_v.popitem()
for local in locals_v:
globals_v[local] = locals_v[local]
exec_result=f(**keywords)
from diskcache import Cache
cache = Cache({path!a})
cache.set({_id!a},{success})
except Exception as e:
from diskcache import Cache
cache = Cache({path!a})
cache.set({_id!a},{err})
"""
if self.sandbox:
subprocess_result = self._exec_sandbox(_exec_code, _id)
else:
subprocess_result = self._exec(_exec_code)
if subprocess_result.returncode == 1:
raise Exception(subprocess_result.stderr)
cache = Cache(self.sandbox_path)
result = cache.get(_id)
cache.delete(_id)
if result.get('code') == 200:
return result.get('data')
raise Exception(result.get('msg'))
def _exec_sandbox(self, _code, _id):
exec_python_file = f'{self.sandbox_path}/{_id}.py'
with open(exec_python_file, 'w') as file:
file.write(_code)
os.system(f"chown {self.user}:{self.user} {exec_python_file}")
kwargs = {'cwd': BASE_DIR}
subprocess_result = subprocess.run(
['su', '-s', python_directory, '-c', "exec(open('" + exec_python_file + "').read())", self.user],
text=True,
capture_output=True, **kwargs)
os.remove(exec_python_file)
return subprocess_result
@staticmethod
def _exec(_code):
return subprocess.run([python_directory, '-c', _code], text=True, capture_output=True)

74
apps/modules/views/module.py
View File

@ -14,13 +14,15 @@ from modules.serializers.module import ModuleSerializer, ModuleTreeSerializer
class ModuleView(APIView):
authentication_classes = [TokenAuth]
@extend_schema(methods=['POST'],
description=_('Create module'),
operation_id=_('Create module'),
parameters=ModuleCreateAPI.get_parameters(),
request=ModuleCreateAPI.get_request(),
responses=ModuleCreateAPI.get_response(),
tags=[_('Module')])
@extend_schema(
methods=['POST'],
description=_('Create module'),
operation_id=_('Create module'),
parameters=ModuleCreateAPI.get_parameters(),
request=ModuleCreateAPI.get_request(),
responses=ModuleCreateAPI.get_response(),
tags=[_('Module')]
)
@has_permissions(lambda r, kwargs: Permission(group=Group(kwargs.get('source')), operate=Operate.CREATE,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}"))
def post(self, request: Request, workspace_id: str, source: str):
@ -30,12 +32,14 @@ class ModuleView(APIView):
'workspace_id': workspace_id}
).insert(request.data))
@extend_schema(methods=['GET'],
description=_('Get module tree'),
operation_id=_('Get module tree'),
parameters=ModuleTreeReadAPI.get_parameters(),
responses=ModuleTreeReadAPI.get_response(),
tags=[_('Module')])
@extend_schema(
methods=['GET'],
description=_('Get module tree'),
operation_id=_('Get module tree'),
parameters=ModuleTreeReadAPI.get_parameters(),
responses=ModuleTreeReadAPI.get_response(),
tags=[_('Module')]
)
@has_permissions(lambda r, kwargs: Permission(group=Group(kwargs.get('source')), operate=Operate.READ,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}"))
def get(self, request: Request, workspace_id: str, source: str):
@ -46,13 +50,15 @@ class ModuleView(APIView):
class Operate(APIView):
authentication_classes = [TokenAuth]
@extend_schema(methods=['PUT'],
description=_('Update module'),
operation_id=_('Update module'),
parameters=ModuleEditAPI.get_parameters(),
request=ModuleEditAPI.get_request(),
responses=ModuleEditAPI.get_response(),
tags=[_('Module')])
@extend_schema(
methods=['PUT'],
description=_('Update module'),
operation_id=_('Update module'),
parameters=ModuleEditAPI.get_parameters(),
request=ModuleEditAPI.get_request(),
responses=ModuleEditAPI.get_response(),
tags=[_('Module')]
)
@has_permissions(lambda r, kwargs: Permission(group=Group(kwargs.get('source')), operate=Operate.EDIT,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}"))
def put(self, request: Request, workspace_id: str, source: str, module_id: str):
@ -60,12 +66,14 @@ class ModuleView(APIView):
data={'id': module_id, 'workspace_id': workspace_id, 'source': source}
).edit(request.data))
@extend_schema(methods=['GET'],
description=_('Get module'),
operation_id=_('Get module'),
parameters=ModuleReadAPI.get_parameters(),
responses=ModuleReadAPI.get_response(),
tags=[_('Module')])
@extend_schema(
methods=['GET'],
description=_('Get module'),
operation_id=_('Get module'),
parameters=ModuleReadAPI.get_parameters(),
responses=ModuleReadAPI.get_response(),
tags=[_('Module')]
)
@has_permissions(lambda r, kwargs: Permission(group=Group(kwargs.get('source')), operate=Operate.READ,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}"))
def get(self, request: Request, workspace_id: str, source: str, module_id: str):
@ -73,12 +81,14 @@ class ModuleView(APIView):
data={'id': module_id, 'workspace_id': workspace_id, 'source': source}
).one())
@extend_schema(methods=['DELETE'],
description=_('Delete module'),
operation_id=_('Delete module'),
parameters=ModuleDeleteAPI.get_parameters(),
responses=ModuleDeleteAPI.get_response(),
tags=[_('Module')])
@extend_schema(
methods=['DELETE'],
description=_('Delete module'),
operation_id=_('Delete module'),
parameters=ModuleDeleteAPI.get_parameters(),
responses=ModuleDeleteAPI.get_response(),
tags=[_('Module')]
)
@has_permissions(lambda r, kwargs: Permission(group=Group(kwargs.get('source')), operate=Operate.DELETE,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}"))
def delete(self, request: Request, workspace_id: str, source: str, module_id: str):

103
apps/tools/api/tool.py
View File

@ -4,7 +4,7 @@ from drf_spectacular.utils import OpenApiParameter
from common.mixins.api_mixin import APIMixin
from common.result import ResultSerializer, DefaultResultSerializer
from tools.serializers.tool import ToolModelSerializer, ToolCreateRequest
from tools.serializers.tool import ToolModelSerializer, ToolCreateRequest, ToolDebugRequest
class ToolCreateResponse(ResultSerializer):
@ -91,3 +91,104 @@ class ToolTreeReadAPI(APIMixin):
required=False,
)
]
class ToolDebugApi(APIMixin):
@staticmethod
def get_request():
return ToolDebugRequest
@staticmethod
def get_response():
return DefaultResultSerializer
class ToolExportAPI(APIMixin):
@staticmethod
def get_parameters():
return [
OpenApiParameter(
name="workspace_id",
description="工作空间id",
type=OpenApiTypes.STR,
location='path',
required=True,
),
OpenApiParameter(
name="tool_id",
description="工具id",
type=OpenApiTypes.STR,
location='path',
required=True,
)
]
@staticmethod
def get_response():
return DefaultResultSerializer
class ToolImportAPI(APIMixin):
@staticmethod
def get_parameters():
return [
OpenApiParameter(
name="workspace_id",
description="工作空间id",
type=OpenApiTypes.STR,
location='path',
required=True,
),
OpenApiParameter(
name='file',
type=OpenApiTypes.BINARY,
description='工具文件',
required=True
),
]
@staticmethod
def get_response():
return DefaultResultSerializer
class ToolPageAPI(ToolReadAPI):
@staticmethod
def get_parameters():
return [
OpenApiParameter(
name="workspace_id",
description="工作空间id",
type=OpenApiTypes.STR,
location='path',
required=True,
),
OpenApiParameter(
name="tool_id",
description="工具id",
type=OpenApiTypes.STR,
location='path',
required=True,
),
OpenApiParameter(
name="current_page",
description="当前页码",
type=OpenApiTypes.INT,
location='path',
required=True,
),
OpenApiParameter(
name="page_size",
description="每页大小",
type=OpenApiTypes.INT,
location='path',
required=True,
),
OpenApiParameter(
name="name",
description="工具名称",
type=OpenApiTypes.STR,
location='query',
required=False,
),
]

171
apps/tools/serializers/tool.py
View File

@ -1,14 +1,57 @@
# -*- coding: utf-8 -*-
import json
import pickle
import re
import uuid_utils.compat as uuid
from django.core import validators
from django.db import transaction
from django.db.models import QuerySet
from django.http import HttpResponse
from django.utils.translation import gettext_lazy as _
from rest_framework import serializers
from rest_framework import serializers, status
from common.exception.app_exception import AppApiException
from common.result import result
from common.utils.tool_code import ToolExecutor
from maxkb.const import CONFIG
from tools.models import Tool, ToolScope, ToolModule
tool_executor = ToolExecutor(CONFIG.get('SANDBOX'))
class ToolInstance:
def __init__(self, tool: dict, version: str):
self.tool = tool
self.version = version
def encryption(message: str):
"""
加密敏感字段数据 加密方式是 如果密码是 1234567890 那么给前端则是 123******890
:param message:
:return:
"""
if type(message) != str:
return message
if message == "":
return ""
max_pre_len = 8
max_post_len = 4
message_len = len(message)
pre_len = int(message_len / 5 * 2)
post_len = int(message_len / 5 * 1)
pre_str = "".join([message[index] for index in
range(0,
max_pre_len if pre_len > max_pre_len else 1 if pre_len <= 0 else int(
pre_len))])
end_str = "".join(
[message[index] for index in
range(message_len - (int(post_len) if pre_len < max_post_len else max_post_len),
message_len)])
content = "***************"
return pre_str + content + end_str
class ToolModelSerializer(serializers.ModelSerializer):
class Meta:
@ -18,6 +61,14 @@ class ToolModelSerializer(serializers.ModelSerializer):
'create_time', 'update_time']
class UploadedFileField(serializers.FileField):
def __init__(self, **kwargs):
super().__init__(**kwargs)
def to_representation(self, value):
return value
class ToolInputField(serializers.Serializer):
name = serializers.CharField(required=True, label=_('variable name'))
is_required = serializers.BooleanField(required=True, label=_('required'))
@ -60,6 +111,20 @@ class ToolCreateRequest(serializers.Serializer):
module_id = serializers.CharField(required=False, allow_null=True, allow_blank=True, default='root')
class DebugField(serializers.Serializer):
name = serializers.CharField(required=True, label=_('variable name'))
value = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_('variable value'))
class ToolDebugRequest(serializers.Serializer):
code = serializers.CharField(required=True, label=_('tool content'))
input_field_list = serializers.ListField(child=ToolInputField(), required=False, default=list,
label=_('input field list'))
init_field_list = serializers.ListField(child=InitField(), required=False, default=list, label=_('init field list'))
init_params = serializers.DictField(required=False, default=dict, label=_('init params'))
debug_field_list = DebugField(required=True, many=True)
class ToolSerializer(serializers.Serializer):
class Create(serializers.Serializer):
user_id = serializers.UUIDField(required=True, label=_('user id'))
@ -82,6 +147,64 @@ class ToolSerializer(serializers.Serializer):
tool.save()
return ToolModelSerializer(tool).data
class Debug(serializers.Serializer):
user_id = serializers.UUIDField(required=True, label=_('user id'))
workspace_id = serializers.CharField(required=True, label=_('workspace id'))
def debug(self, debug_instance):
self.is_valid(raise_exception=True)
ToolDebugRequest(data=debug_instance).is_valid(raise_exception=True)
input_field_list = debug_instance.get('input_field_list')
code = debug_instance.get('code')
debug_field_list = debug_instance.get('debug_field_list')
init_params = debug_instance.get('init_params')
params = {field.get('name'): self.convert_value(field.get('name'), field.get('value'), field.get('type'),
field.get('is_required'))
for field in
[{'value': self.get_field_value(debug_field_list, field.get('name'), field.get('is_required')),
**field} for field in
input_field_list]}
# 合并初始化参数
if init_params is not None:
all_params = init_params | params
else:
all_params = params
return tool_executor.exec_code(code, all_params)
@staticmethod
def get_field_value(debug_field_list, name, is_required):
result = [field for field in debug_field_list if field.get('name') == name]
if len(result) > 0:
return result[-1].get('value')
if is_required:
raise AppApiException(500, f"{name}" + _('field has no value set'))
return None
@staticmethod
def convert_value(name: str, value: str, _type: str, is_required: bool):
if not is_required and value is None:
return None
try:
if _type == 'int':
return int(value)
if _type == 'float':
return float(value)
if _type == 'dict':
v = json.loads(value)
if isinstance(v, dict):
return v
raise Exception(_('type error'))
if _type == 'array':
v = json.loads(value)
if isinstance(v, list):
return v
raise Exception(_('type error'))
return value
except Exception as e:
raise AppApiException(500, _('Field: {name} Type: {_type} Value: {value} Type conversion error').format(
name=name, type=_type, value=value
))
class Operate(serializers.Serializer):
id = serializers.UUIDField(required=True, label=_('tool id'))
workspace_id = serializers.CharField(required=True, label=_('workspace id'))
@ -111,6 +234,52 @@ class ToolSerializer(serializers.Serializer):
tool = QuerySet(Tool).filter(id=self.data.get('id')).first()
return ToolModelSerializer(tool).data
def export(self):
try:
self.is_valid()
id = self.data.get('id')
tool = QuerySet(Tool).filter(id=id).first()
tool_dict = ToolModelSerializer(tool).data
mk_instance = ToolInstance(tool_dict, 'v2')
tool_pickle = pickle.dumps(mk_instance)
response = HttpResponse(content_type='text/plain', content=tool_pickle)
response['Content-Disposition'] = f'attachment; filename="{tool.name}.fx"'
return response
except Exception as e:
return result.error(str(e), response_status=status.HTTP_500_INTERNAL_SERVER_ERROR)
class Import(serializers.Serializer):
file = UploadedFileField(required=True, label=_("file"))
user_id = serializers.UUIDField(required=True, label=_("User ID"))
workspace_id = serializers.CharField(required=True, label=_("workspace id"))
#
@transaction.atomic
def import_(self):
self.is_valid()
# user_id = self.data.get('user_id')
# flib_instance_bytes = self.data.get('file').read()
# try:
# RestrictedUnpickler(io.BytesIO(s)).load()
# flib_instance = restricted_loads(flib_instance_bytes)
# except Exception as e:
# raise AppApiException(1001, _("Unsupported file format"))
# tool = flib_instance.tool
# tool_model = Tool(
# id=uuid.uuid7(),
# name=tool.get('name'),
# desc=tool.get('desc'),
# code=tool.get('code'),
# user_id=user_id,
# input_field_list=tool.get('input_field_list'),
# init_field_list=tool.get('init_field_list', []),
# scope=ToolScope.WORKSPACE,
# is_active=False
# )
# tool_model.save()
return True
class ToolTreeSerializer(serializers.Serializer):
workspace_id = serializers.CharField(required=True, label=_('workspace id'))

4
apps/tools/urls.py
View File

@ -5,5 +5,9 @@ from . import views
app_name = "tool"
urlpatterns = [
path('workspace/<str:workspace_id>/tool', views.ToolView.as_view()),
path('workspace/<str:workspace_id>/tool/import', views.ToolView.Import.as_view()),
path('workspace/<str:workspace_id>/tool/<str:tool_id>', views.ToolView.Operate.as_view()),
path('workspace/<str:workspace_id>/tool/<str:tool_id>/debug', views.ToolView.Debug.as_view()),
path('workspace/<str:workspace_id>/tool/<str:tool_id>/export', views.ToolView.Export.as_view()),
path('workspace/<str:workspace_id>/tool/<int:current_page>/<int:page_size>', views.ToolView.Page.as_view()),
]

153
apps/tools/views/tool.py
View File

@ -1,5 +1,6 @@
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.parsers import MultiPartParser
from rest_framework.request import Request
from rest_framework.views import APIView
@ -7,76 +8,162 @@ from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import PermissionConstants
from common.result import result
from tools.api.tool import ToolCreateAPI, ToolEditAPI, ToolReadAPI, ToolDeleteAPI, ToolTreeReadAPI
from tools.api.tool import ToolCreateAPI, ToolEditAPI, ToolReadAPI, ToolDeleteAPI, ToolTreeReadAPI, ToolDebugApi, \
ToolExportAPI, ToolImportAPI, ToolPageAPI
from tools.serializers.tool import ToolSerializer, ToolTreeSerializer
class ToolView(APIView):
authentication_classes = [TokenAuth]
@extend_schema(methods=['POST'],
description=_('Create tool'),
operation_id=_('Create tool'),
parameters=ToolCreateAPI.get_parameters(),
request=ToolCreateAPI.get_request(),
responses=ToolCreateAPI.get_response(),
tags=[_('Tool')])
@extend_schema(
methods=['POST'],
description=_('Create tool'),
operation_id=_('Create tool'),
parameters=ToolCreateAPI.get_parameters(),
request=ToolCreateAPI.get_request(),
responses=ToolCreateAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_CREATE.get_workspace_permission())
def post(self, request: Request, workspace_id: str):
return result.success(ToolSerializer.Create(
data={'user_id': request.user.id, 'workspace_id': workspace_id}
).insert(request.data))
@extend_schema(methods=['GET'],
description=_('Get tool by module'),
operation_id=_('Get tool by module'),
parameters=ToolTreeReadAPI.get_parameters(),
responses=ToolTreeReadAPI.get_response(),
tags=[_('Tool')])
@extend_schema(
methods=['GET'],
description=_('Get tool by module'),
operation_id=_('Get tool by module'),
parameters=ToolTreeReadAPI.get_parameters(),
responses=ToolTreeReadAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_READ.get_workspace_permission())
def get(self, request: Request, workspace_id: str):
return result.success(ToolTreeSerializer(
data={'workspace_id': workspace_id}
).get_tools(request.query_params.get('module_id')))
class Debug(APIView):
authentication_classes = [TokenAuth]
@extend_schema(
methods=['POST'],
description=_('Debug Tool'),
operation_id=_('Debug Tool'),
request=ToolDebugApi.get_request(),
responses=ToolDebugApi.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_DEBUG.get_workspace_permission())
def post(self, request: Request, workspace_id: str, tool_id: str):
return result.success(ToolSerializer.Debug(
data={'tool_id': tool_id, 'workspace_id': workspace_id}
).debug(request.data))
class Operate(APIView):
authentication_classes = [TokenAuth]
@extend_schema(methods=['PUT'],
description=_('Update tool'),
operation_id=_('Update tool'),
parameters=ToolEditAPI.get_parameters(),
request=ToolEditAPI.get_request(),
responses=ToolEditAPI.get_response(),
tags=[_('Tool')])
@extend_schema(
methods=['PUT'],
description=_('Update tool'),
operation_id=_('Update tool'),
parameters=ToolEditAPI.get_parameters(),
request=ToolEditAPI.get_request(),
responses=ToolEditAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_EDIT.get_workspace_permission())
def put(self, request: Request, workspace_id: str, tool_id: str):
return result.success(ToolSerializer.Operate(
data={'id': tool_id, 'workspace_id': workspace_id}
).edit(request.data))
@extend_schema(methods=['GET'],
description=_('Get tool'),
operation_id=_('Get tool'),
parameters=ToolReadAPI.get_parameters(),
responses=ToolReadAPI.get_response(),
tags=[_('Tool')])
@extend_schema(
methods=['GET'],
description=_('Get tool'),
operation_id=_('Get tool'),
parameters=ToolReadAPI.get_parameters(),
responses=ToolReadAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_READ.get_workspace_permission())
def get(self, request: Request, workspace_id: str, tool_id: str):
return result.success(ToolSerializer.Operate(
data={'id': tool_id, 'workspace_id': workspace_id}
).one())
@extend_schema(methods=['DELETE'],
description=_('Delete tool'),
operation_id=_('Delete tool'),
parameters=ToolDeleteAPI.get_parameters(),
responses=ToolDeleteAPI.get_response(),
tags=[_('Tool')])
@extend_schema(
methods=['DELETE'],
description=_('Delete tool'),
operation_id=_('Delete tool'),
parameters=ToolDeleteAPI.get_parameters(),
responses=ToolDeleteAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_DELETE.get_workspace_permission())
def delete(self, request: Request, workspace_id: str, tool_id: str):
return result.success(ToolSerializer.Operate(
data={'id': tool_id, 'workspace_id': workspace_id}
).delete())
class Page(APIView):
authentication_classes = [TokenAuth]
@extend_schema(
methods=['GET'],
description=_('Get tool list by pagination'),
operation_id=_('Get tool list by pagination'),
parameters=ToolPageAPI.get_parameters(),
responses=ToolPageAPI.get_response(),
tags=[_('Tool')]
)
@has_permissions(PermissionConstants.TOOL_READ.get_workspace_permission())
def get(self, request: Request, current_page: int, page_size: int):
return result.success(
ToolSerializer.Query(
data={
'name': request.query_params.get('name'),
'desc': request.query_params.get('desc'),
'function_type': request.query_params.get('function_type'),
'user_id': request.user.id,
'select_user_id': request.query_params.get('select_user_id')
}
).page(current_page, page_size))
class Import(APIView):
authentication_classes = [TokenAuth]
parser_classes = [MultiPartParser]
@extend_schema(
methods=['POST'],
description=_("Import tool"),
operation_id=_("Import tool"),
parameters=ToolImportAPI.get_parameters(),
request=ToolImportAPI.get_request(),
responses=ToolImportAPI.get_response(),
tags=[_("Tool")]
)
@has_permissions(PermissionConstants.TOOL_IMPORT.get_workspace_permission())
def post(self, request: Request, workspace_id: str):
return result.success(ToolSerializer.Import(
data={'workspace_id': workspace_id, 'file': request.FILES.get('file'), 'user_id': request.user.id}
).import_())
class Export(APIView):
authentication_classes = [TokenAuth]
@extend_schema(
methods=['GET'],
description=_("Export tool"),
operation_id=_("Export tool"),
parameters=ToolExportAPI.get_parameters(),
responses=ToolExportAPI.get_response(),
tags=[_("Tool")]
)
@has_permissions(PermissionConstants.TOOL_EXPORT.get_workspace_permission())
def get(self, request: Request, tool_id: str, workspace_id: str):
return ToolSerializer.Operate(
data={'id': tool_id, 'workspace_id': workspace_id}
).export()