Merge pull request #128 from TencentBlueKing/develop

Merge dev to master for release V1.9.3

Author: normal-wls

app_desc.yaml

@@ -1,5 +1,5 @@
 spec_version: 2
-app_version: "1.9.2"
+app_version: "1.9.3"
 app:
   region: default
   bk_app_code: &APP_CODE bk_flow_engine

bkflow/pipeline_plugins/components/collections/uniform_api/v2_0_0.py

@@ -34,6 +34,7 @@
 )
 from bkflow.pipeline_plugins.query.uniform_api.utils import UniformAPIClient
 from bkflow.pipeline_plugins.utils import convert_dict_value
+from bkflow.space.configs import UniformAPIConfigHandler
 from bkflow.utils.api_client import HttpRequestResult
 from bkflow.utils.handlers import handle_plain_log
 
@@ -122,7 +123,6 @@ def _dispatch_schedule_trigger(self, data, parent_data, callback_data=None):
         callback = api_data.pop("uniform_api_plugin_callback", None)
         method = api_data.pop("uniform_api_plugin_method")
         resp_data_path: str = api_data.pop("response_data_path", None)
-
         # 获取空间相关配置信息
         interface_client = InterfaceModuleClient()
         space_infos_result = interface_client.get_space_infos(
@@ -137,12 +137,18 @@ def _dispatch_schedule_trigger(self, data, parent_data, callback_data=None):
             return False
 
         space_configs = space_infos_result.get("data", {}).get("configs", {})
+        uniform_api_config = space_configs.get("uniform_api", {})
+        validated_config = UniformAPIConfigHandler(uniform_api_config).handle()
+        if validated_config.exclude_none_fields:
+            # 过滤字符串为空的基础类型
+            keys_to_remove = [key for key, value in api_data.items() if value == ""]
+            self.logger.info(f"none fields keys to remove: {keys_to_remove}")
+            for key in keys_to_remove:
+                api_data.pop(key)
+            self.logger.info(f"plugin_data after poping: {api_data}")
 
         # 开启的enable_api_parameter_conversion配置只对POST参数生效
-        if (
-            space_configs.get("uniform_api", {}).get("enable_api_parameter_conversion", False)
-            and method.upper() == "POST"
-        ):
+        if validated_config.enable_api_parameter_conversion:
             # 启动参数转换
             api_data = convert_dict_value(api_data)
 

bkflow/pipeline_plugins/query/uniform_api/uniform_api.py

@@ -24,7 +24,7 @@
 from bkflow.exceptions import APIResponseError, ValidationError
 from bkflow.pipeline_plugins.query.uniform_api.utils import UniformAPIClient
 from bkflow.pipeline_plugins.query.utils import query_response_handler
-from bkflow.space.configs import UniformApiConfig
+from bkflow.space.configs import UniformApiConfig, UniformAPIConfigHandler
 from bkflow.space.models import SpaceConfig
 from bkflow.utils.api_client import HttpRequestResult
 
@@ -33,6 +33,7 @@ class UniformAPICategorySerializer(serializers.Serializer):
     scope_type = serializers.CharField(required=False)
     scope_value = serializers.CharField(required=False)
     key = serializers.CharField(required=False)
+    api_name = serializers.CharField(required=False)
 
 
 class UniformAPIListSerializer(serializers.Serializer):
@@ -42,6 +43,7 @@ class UniformAPIListSerializer(serializers.Serializer):
     scope_value = serializers.CharField(required=False)
     category = serializers.CharField(required=False)
     key = serializers.CharField(required=False)
+    api_name = serializers.CharField(required=False)
 
 
 class UniformAPIMetaSerializer(serializers.Serializer):
@@ -52,10 +54,15 @@ class UniformAPIMetaSerializer(serializers.Serializer):
 
 def _get_space_uniform_api_list_info(space_id, request_data, config_key):
     uniform_api_config = SpaceConfig.get_config(space_id=space_id, config_name=UniformApiConfig.name)
-    if not uniform_api_config.get(config_key):
+    if not uniform_api_config:
         raise ValidationError("接入平台未注册统一API, 请联系对应接入平台管理员")
     client = UniformAPIClient()
-    url = uniform_api_config[config_key]
+    uniform_api_config = UniformAPIConfigHandler(uniform_api_config).handle()
+    # 弹出此参数避免透传
+    api_name = request_data.pop("api_name", UniformApiConfig.Keys.DEFAULT_API_KEY.value)
+    url = uniform_api_config.api.get(api_name, {}).get(config_key)
+    if not url:
+        raise ValidationError("对应API未配置, 请联系对应接入平台管理员")
     request_result: HttpRequestResult = client.request(url=url, method="GET", data=request_data)
     if not request_result.result:
         raise APIResponseError(f"请求统一API列表失败: {request_result.message}")

bkflow/space/configs.py

@@ -18,16 +18,19 @@
 to the current version of the project delivered to anyone in the future.
 """
 from enum import Enum
-from typing import Type
+from typing import Dict, Optional, Type
 
 import jsonschema
 from django.utils.translation import ugettext_lazy as _
+from pydantic import BaseModel, constr
 from pytimeparse import parse
 
 from bkflow.exceptions import ValidationError
 from bkflow.plugin.space_plugin_config_parser import SpacePluginConfigParser
 from bkflow.utils.apigw import check_url_from_apigw
 
+valid_api_key = constr(regex=r"^[A-Za-z0-9_]+$")
+
 
 class SpaceConfigValueType(Enum):
     # json 类型
@@ -98,6 +101,7 @@ def get_config(cls, name):
 
     @classmethod
     def get_all_configs(cls, only_public=False):
+
         # copy, 降低被修改风险
         if only_public:
             return {name: config_cls for name, config_cls in cls.__hub.items() if config_cls.is_public}
@@ -203,32 +207,33 @@ def validate(cls, value: dict):
 
 class UniformApiConfig(BaseSpaceConfig):
     name = "uniform_api"
-    desc = _("是否开启统一API")
     value_type = SpaceConfigValueType.JSON.value
     default_value = {}
-    example = {"meta_apis": "{meta_apis url}", "api_categories": "{api_categories url}"}
+    example = {
+        "api": {
+            "{api_key}": {
+                "meta_apis": "{meta_apis url}",
+                "api_categories": "{api_categories url}",
+                "display_name": "{display_name}",
+            }
+        }
+    }
+    desc = _("API 插件配置 （如更改配置，可能对已存在数据产生不兼容影响，请谨慎操作）")
+    """
+    仍然支持读取 旧 SCHEMA 但不能支持继续配置
+    旧 SCHEMA 格式 example = {"meta_apis": "{meta_apis url}", "api_categories": "{api_categories url}"}
+    """
 
     class Keys(Enum):
         META_APIS = "meta_apis"
         API_CATEGORIES = "api_categories"
-
-    SCHEMA = {
-        "type": "object",
-        "required": ["meta_apis"],
-        "properties": {
-            Keys.META_APIS.value: {"type": "string"},
-            Keys.API_CATEGORIES.value: {"type": "string"},
-        },
-    }
+        DISPLAY_NAME = "display_name"
+        DEFAULT_DISPLAY_NAME = "API插件"
+        DEFAULT_API_KEY = "default"
 
     @classmethod
-    def validate(cls, value: dict):
-        try:
-            jsonschema.validate(value, cls.SCHEMA)
-        except jsonschema.ValidationError as e:
-            raise ValidationError(f"[validate uniform api config error]: {str(e)}")
-
-        meta_apis_from_apigw = check_url_from_apigw(value[cls.Keys.META_APIS.value])
+    def check_url(cls, value):
+        meta_apis_from_apigw = check_url_from_apigw(value.get(cls.Keys.META_APIS.value))
         category_config = value.get(cls.Keys.API_CATEGORIES.value)
         api_categories_from_apigw = check_url_from_apigw(category_config) if category_config else True
         if not (api_categories_from_apigw and meta_apis_from_apigw):
@@ -237,6 +242,16 @@ def validate(cls, value: dict):
             )
         return True
 
+    @classmethod
+    def validate(cls, value: dict):
+        try:
+            model = SchemaV2Model(**value)
+        except ValueError as e:
+            raise ValidationError(f"[validate uniform api config error]: {str(e)} should have {str(cls.example)}")
+        for obj in model.api.values():
+            cls.check_url(obj)
+        return True
+
 
 class SuperusersConfig(BaseSpaceConfig):
     name = "superusers"
@@ -297,3 +312,69 @@ class SpacePluginConfig(BaseSpaceConfig):
     @classmethod
     def validate(cls, value: dict):
         return SpacePluginConfigParser(config=value).is_valid()
+
+
+# 定义 SCHEMA_V1 对应的模型
+class SchemaV1Model(BaseModel):
+    meta_apis: str
+    api_categories: Optional[str] = None
+
+
+# 定义 SCHEMA 对应的模型
+class ApiModel(BaseModel):
+    meta_apis: str
+    api_categories: str
+    display_name: str
+
+    def get(self, field_name, default=None):
+        # 由于获取插件种类/列表时候传入的 key 不确定 需要提供一个 get 方法
+        return getattr(self, field_name, default)
+
+
+class CommonModel(BaseModel):
+    exclude_none_fields: Optional[str] = None
+    enable_api_parameter_conversion: Optional[str] = None
+
+
+class SchemaV2Model(BaseModel):
+    api: Dict[valid_api_key, ApiModel]
+    common: Optional[CommonModel] = None
+
+    def __getattr__(self, key):
+        try:
+            super().__getattribute__(key)
+        except AttributeError:
+            # 当前没有则从 common 中获取 如果出现不存在的字段则报错
+            if key not in CommonModel.__fields__:
+                raise
+            if self.common and hasattr(self.common, key):
+                return getattr(self.common, key)
+            return None
+
+
+class UniformAPIConfigHandler:
+    def __init__(self, config: dict):
+        self.config = config
+
+    def handle(self):
+        model = None
+        try:
+            # 尝试按新协议解析
+            model = SchemaV2Model(**self.config)
+            return model
+        except ValueError:
+            pass
+        try:
+            # 兼容旧协议解析
+            v1_model = SchemaV1Model(**self.config)
+        except ValueError as e:
+            raise ValidationError(
+                f"[validate uniform api config error]: {str(e)} should have {UniformApiConfig.example}"
+            )
+        api_model = ApiModel(
+            meta_apis=v1_model.meta_apis,
+            api_categories=v1_model.api_categories,
+            display_name=UniformApiConfig.Keys.DEFAULT_DISPLAY_NAME.value,
+        )
+        model = SchemaV2Model(api={UniformApiConfig.Keys.DEFAULT_API_KEY.value: api_model})
+        return model

bkflow/space/views.py

@@ -196,9 +196,7 @@ def get_space_infos(self, request, *args, **kwargs):
                         space_id=data["space_id"], name=api_gateway_credential_name, type=CredentialType.BK_APP.value
                     ).value
                 except (Credential.DoesNotExist, SpaceConfigDefaultValueNotExists) as e:
-                    logger.exception(
-                        "CredentialViewSet 获取空间下的凭证异常, space_id={}, err={}, ".format(data["space_id"], e)
-                    )
+                    logger.exception("CredentialViewSet 获取空间下的凭证异常, space_id={}, err={}, ".format(data["space_id"], e))
                     value = {}
             else:
                 value = SpaceConfig.get_config(space_id=data["space_id"], config_name=config_name)
@@ -229,13 +227,16 @@ def list(self, request, *args, **kwargs):
             raise PermissionDenied()
         return super().list(request, *args, **kwargs)
 
-    @swagger_auto_schema(
-        method="get", operation_summary="获取所有空间配置元信息", query_serializer=SpaceConfigBaseQuerySerializer
-    )
+    def process_config(self, config_dict):
+        if not config_dict.get("default_value"):
+            config_dict["default_value"] = None
+        return config_dict
+
+    @swagger_auto_schema(method="get", operation_summary="获取所有空间配置元信息", query_serializer=SpaceConfigBaseQuerySerializer)
     @action(detail=False, methods=["GET"])
     def config_meta(self, request, *args, **kwargs):
         configs = SpaceConfigHandler.get_all_configs()
-        return Response({name: config.to_dict() for name, config in configs.items()})
+        return Response({name: self.process_config(config.to_dict()) for name, config in configs.items()})
 
     @swagger_auto_schema(
         method="post",
@@ -250,9 +251,7 @@ def batch_apply(self, request, *args, **kwargs):
         SpaceConfig.objects.batch_update(space_id=space_id, configs=configs)
         return Response(SpaceConfig.objects.get_space_config_info(space_id=space_id, simplified=False))
 
-    @swagger_auto_schema(
-        method="get", operation_summary="获取空间下所有配置", query_serializer=SpaceConfigBaseQuerySerializer
-    )
+    @swagger_auto_schema(method="get", operation_summary="获取空间下所有配置", query_serializer=SpaceConfigBaseQuerySerializer)
     @action(detail=False, methods=["GET"])
     def get_all_space_configs(self, request, *args, **kwargs):
         ser = SpaceConfigBaseQuerySerializer(data=request.query_params)

bkflow/task/node_log.py

@@ -59,7 +59,7 @@ def fetch_node_logs(self, node_id, version_id, *args, **kwargs):
         if self.private_token:
             url_params.update({"private_token": self.private_token})
         url = self.url.rstrip("/") + f"/?{urlencode(url_params)}"
-        payload = {"query": {"query_string": f"json.node_id:{node_id} AND json.version:{version_id}"}}
+        payload = {"query": {"query_string": f"__ext_json.node_id:{node_id} AND __ext_json.version:{version_id}"}}
         response = requests.get(url, headers=self.headers, data=json.dumps(payload))
         logger.info(
             f"[PaaS3NodeLogDataSource fetch_node_logs] request {url} with payload {payload} and "

bkflow/template/urls.py

@@ -26,6 +26,7 @@
     TemplateMockDataViewSet,
     TemplateMockSchemeViewSet,
     TemplateViewSet,
+    TemplateMockTaskViewSet,
 )
 from bkflow.template.views.variable import VariableViewSet
 
@@ -34,6 +35,7 @@
 router.register(r"^admin", AdminTemplateViewSet, basename="admin_template")
 router.register(r"^template_mock_data", TemplateMockDataViewSet, basename="template_mock_data")
 router.register(r"^template_mock_scheme", TemplateMockSchemeViewSet, basename="template_mock_scheme")
+router.register(r"^template_mock_task", TemplateMockTaskViewSet, basename="template_mock_task")
 router.register(r"", TemplateViewSet, basename="template")
 
 urlpatterns = [

bkflow/template/views/template.py

@@ -26,6 +26,7 @@
 from rest_framework import mixins
 from rest_framework.decorators import action
 from rest_framework.response import Response
+from rest_framework.viewsets import GenericViewSet
 
 from bkflow.apigw.serializers.task import (
     CreateMockTaskWithPipelineTreeSerializer,
@@ -40,7 +41,11 @@
 from bkflow.pipeline_web.drawing_new.drawing import draw_pipeline as draw_pipeline_tree
 from bkflow.pipeline_web.preview import preview_template_tree
 from bkflow.pipeline_web.preview_base import PipelineTemplateWebPreviewer
-from bkflow.space.configs import GatewayExpressionConfig
+from bkflow.space.configs import (
+    GatewayExpressionConfig,
+    UniformApiConfig,
+    UniformAPIConfigHandler,
+)
 from bkflow.space.exceptions import SpaceConfigDefaultValueNotExists
 from bkflow.space.models import SpaceConfig
 from bkflow.space.permissions import SpaceSuperuserPermission
@@ -69,6 +74,7 @@
     TemplateMockDataSerializer,
     TemplateMockSchemeSerializer,
     TemplateOperationRecordSerializer,
+    TemplateRelatedResourceSerializer,
     TemplateSerializer,
 )
 from bkflow.template.utils import analysis_pipeline_constants_ref
@@ -155,9 +161,7 @@ def create_task(self, request, space_id, *args, **kwargs):
             raise APIResponseError(result["message"])
         return Response(result["data"])
 
-    @swagger_auto_schema(
-        method="POST", operation_description="流程批量删除", request_body=TemplateBatchDeleteSerializer
-    )
+    @swagger_auto_schema(method="POST", operation_description="流程批量删除", request_body=TemplateBatchDeleteSerializer)
     @action(methods=["POST"], detail=False, url_path="batch_delete")
     def batch_delete(self, request, *args, **kwargs):
         ser = TemplateBatchDeleteSerializer(data=request.data)
@@ -255,6 +259,11 @@ def get_space_related_configs(self, request, *args, **kwargs):
         except SpaceConfigDefaultValueNotExists as e:
             logger.error(f"space_id={template.space_id}, config_name={GatewayExpressionConfig.name}, error={e}")
             raise
+        # 增加 uniform api 的配置项返回
+        uniform_api_config = SpaceConfig.get_config(space_id=template.space_id, config_name=UniformApiConfig.name)
+        if uniform_api_config:
+            uniform_api_config = UniformAPIConfigHandler(uniform_api_config).handle().dict()
+        data.update({UniformApiConfig.name: uniform_api_config})
         return Response(data)
 
     @swagger_auto_schema(methods=["post"], operation_description="流程树预览", request_body=PreviewTaskTreeSerializer)
@@ -419,3 +428,17 @@ def perform_create(self, serializer):
     def perform_update(self, serializer):
         user = serializer.context.get("request").user
         serializer.save(operator=user.username)
+
+
+class TemplateMockTaskViewSet(mixins.ListModelMixin, GenericViewSet):
+    DEFAULT_PERMISSION = TemplateRelatedResourcePermission.MOCK_PERMISSION
+    permission_classes = [AdminPermission | SpaceSuperuserPermission | TemplateRelatedResourcePermission]
+
+    @swagger_auto_schema(query_serializer=TemplateRelatedResourceSerializer)
+    def list(self, request, *args, **kwargs):
+        ser = TemplateRelatedResourceSerializer(data=request.query_params)
+        ser.is_valid(raise_exception=True)
+        space_id, template_id = ser.validated_data["space_id"], ser.validated_data["template_id"]
+        client = TaskComponentClient(space_id=space_id)
+        result = client.task_list(data={"template_id": template_id, "space_id": space_id, "create_method": "MOCK"})
+        return Response(result)