From 042a71bdbe3c599b7b60d53a7dc79d8f73c5282a Mon Sep 17 00:00:00 2001 From: jpyoung3 <809608046@qq.com> Date: Fri, 13 Dec 2024 11:32:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=89=E8=A3=85=E9=A2=84=E8=AE=BE?= =?UTF-8?q?=E6=8F=92=E4=BB=B6=E9=94=81=E5=AE=9A=E7=89=88=E6=9C=AC=20(close?= =?UTF-8?q?d=20#2482)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/backend/subscription/steps/adapter.py | 30 ++++++++++++---------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/apps/backend/subscription/steps/adapter.py b/apps/backend/subscription/steps/adapter.py index 6816da833..1b324472b 100644 --- a/apps/backend/subscription/steps/adapter.py +++ b/apps/backend/subscription/steps/adapter.py @@ -285,19 +285,19 @@ def max_ids_by_key(self, contained_os_cpu_items: List[Dict[str, Any]]) -> List[i def format2policy_packages_new( self, plugin_id: int, plugin_name: str, plugin_version: str, config_templates: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: - latest_flag: str = "latest" - is_tag: bool = Tag.objects.filter( - target_id=plugin_id, name=latest_flag, target_type=TargetType.PLUGIN.value - ).exists() + tags: List[str] = ["latest", "stable"] + is_tag: Dict[str, bool] = { + tag: Tag.objects.filter(target_id=plugin_id, name=tag, target_type=TargetType.PLUGIN.value).exists() + for tag in tags + } - if plugin_version != latest_flag or is_tag: - # 如果 latest 是 tag，走取指定版本的逻辑 - packages = self.get_packages(plugin_name, plugin_version) - else: + if plugin_version in tags and not is_tag[plugin_version]: max_pkg_ids: List[int] = self.max_ids_by_key( list(models.Packages.objects.filter(project=plugin_name).values("id", "os", "cpu_arch")) ) packages = models.Packages.objects.filter(id__in=max_pkg_ids) + else: + packages = self.get_packages(plugin_name, plugin_version) if not packages: raise errors.PluginValidationError( @@ -307,11 +307,11 @@ def format2policy_packages_new( os_cpu__config_templates_map = defaultdict(list) for template in config_templates: is_main_template = template["is_main"] - if template["version"] != latest_flag or is_tag: - plugin_version_set = {plugin_version, "*"} + if plugin_version in tags and not is_tag[plugin_version]: + tag_packages_version_set = set(packages.values_list("version", flat=True)) + plugin_version_set = tag_packages_version_set | {"*"} else: - latest_packages_version_set = set(packages.values_list("version", flat=True)) - plugin_version_set = latest_packages_version_set | {"*"} + plugin_version_set = {plugin_version, "*"} max_config_tmpl_ids: typing.List[int] = self.max_ids_by_key( list( @@ -479,6 +479,7 @@ def get_matching_package_obj(self, os_type: str, cpu_arch: str, bk_biz_id: int) def get_matching_config_tmpl_objs( self, os_type: str, cpu_arch: str, package: models.Packages, bk_biz_id: int, config: Dict ) -> List[models.PluginConfigTemplate]: + # 如果当前业务设定了业务最大版本，重新从数据库中获取 if str(bk_biz_id) in self.plugin_version_config(): config_tmpl = ( models.PluginConfigTemplate.objects.filter( @@ -494,6 +495,7 @@ def get_matching_config_tmpl_objs( return self.config_tmpl_obj_gby_os_key.get(self.get_os_key(os_type, cpu_arch), []) def get_biz_package(self, plugin_name: str, os_type: str, cpu_arch: str, biz_version: str): + """获取业务锁定版本的插件包""" packages_all = self.get_packages(plugin_name, biz_version) packages = packages_all.filter( id__in=[pkg.id for pkg in packages_all if version.Version(pkg.version) <= version.Version(biz_version)] @@ -508,6 +510,7 @@ def get_biz_package(self, plugin_name: str, os_type: str, cpu_arch: str, biz_ver @staticmethod def plugin_version_config(): + """业务锁定版本配置""" plugin_version_config: Dict[str, Dict[str, str]] = models.GlobalSettings.get_config( models.GlobalSettings.KeyEnum.PLUGIN_VERSION_CONFIG.value, default={} ) @@ -515,6 +518,7 @@ def plugin_version_config(): @staticmethod def get_biz_version(package: models.Packages, bk_biz_id: int): + """获取业务锁定版本""" plugin_version_config = PolicyStepAdapter.plugin_version_config() biz_version = None if str(bk_biz_id) in plugin_version_config: @@ -530,7 +534,7 @@ def get_biz_version(package: models.Packages, bk_biz_id: int): return biz_version def get_packages(self, plugin_name: str, plugin_version: str, biz_version: str = None): - """如果不存在某个版本，则返回最大id的版本""" + """如果不存在某个版本，则取最大id的版本""" all_packages = models.Packages.objects.filter(project=plugin_name).values("id", "os", "cpu_arch", "version") version_packages = {pkg["id"]: pkg for pkg in all_packages if pkg["version"] == plugin_version} package_ids = set(version_packages.keys())