Merge branch 'main' into chore/update-plan-code-smells

Author: jimisola

src/reqstool/common/filter_parser.py

@@ -0,0 +1,78 @@
+# Copyright © LFV
+
+from typing import Callable, Dict, Optional, Set, Type, TypeVar
+
+from reqstool.common.models.urn_id import UrnId
+from reqstool.common.utils import Utils
+from reqstool.filters.id_filters import IDFilters
+
+F = TypeVar("F", bound=IDFilters)
+
+
+def parse_filters(
+    data: dict,
+    ids_key: str,
+    filter_cls: Type[F],
+    validate_fn: Callable[[dict], None],
+) -> Dict[str, F]:
+    """Parse filter entries from raw YAML data into a dict of filter objects.
+
+    This utility extracts the duplicated filter-parsing logic that was previously
+    present in both RequirementsModelGenerator and SVCsModelGenerator.
+
+    Args:
+        data: Raw YAML data dict loaded from requirements.yml or svcs.yml.
+        ids_key: The key used for explicit ID filters, either ``"requirement_ids"``
+            or ``"svc_ids"``.
+        filter_cls: The filter dataclass to instantiate, either
+            :class:`~reqstool.filters.requirements_filters.RequirementFilter` or
+            :class:`~reqstool.filters.svcs_filters.SVCFilter`.
+        validate_fn: A callable that performs semantic validation on ``data`` (e.g.
+            checks that includes/excludes are mutually exclusive). Called once before
+            any parsing begins.
+
+    Returns:
+        A dict mapping URN strings to filter instances. Returns an empty dict when
+        ``data`` contains no ``"filters"`` section.
+    """
+    r_filters: Dict[str, F] = {}
+
+    validate_fn(data)
+
+    if "filters" not in data:
+        return r_filters
+
+    for urn in data["filters"].keys():
+        urn_filter = data["filters"][urn]
+
+        urn_ids_imports: Optional[Set[UrnId]] = None
+        urn_ids_excludes: Optional[Set[UrnId]] = None
+        custom_includes = None
+        custom_exclude = None
+
+        if ids_key in urn_filter:
+            if "includes" in urn_filter[ids_key]:
+                filtered_ids = Utils.check_ids_to_filter(current_urn=urn, ids=urn_filter[ids_key]["includes"])
+                ids_includes = list(filtered_ids)
+                urn_ids_imports = set(Utils.convert_ids_to_urn_id(urn=urn, ids=ids_includes))
+
+            if "excludes" in urn_filter[ids_key]:
+                filtered_ids = Utils.check_ids_to_filter(current_urn=urn, ids=urn_filter[ids_key]["excludes"])
+                ids_excludes = list(filtered_ids)
+                urn_ids_excludes = set(Utils.convert_ids_to_urn_id(urn=urn, ids=ids_excludes))
+
+        if "custom" in urn_filter:
+            if "includes" in urn_filter["custom"]:
+                custom_includes = urn_filter["custom"]["includes"]
+
+            if "excludes" in urn_filter["custom"]:
+                custom_exclude = urn_filter["custom"]["excludes"]
+
+        r_filters[urn] = filter_cls(
+            urn_ids_imports=urn_ids_imports,
+            urn_ids_excludes=urn_ids_excludes,
+            custom_imports=custom_includes,
+            custom_exclude=custom_exclude,
+        )
+
+    return r_filters

src/reqstool/model_generators/requirements_model_generator.py

@@ -3,12 +3,13 @@
 import re
 import sys
 from enum import Enum, unique
-from typing import Dict, List, Optional, Set
+from typing import Dict, List
 
 from reqstool_python_decorators.decorators.decorators import Requirements
 from ruamel.yaml import YAML
 
 from reqstool.commands.exit_codes import EXIT_CODE_SYNTAX_VALIDATION_ERROR
+from reqstool.common.filter_parser import parse_filters
 from reqstool.common.models.lifecycle import LifecycleData
 from reqstool.common.models.urn_id import UrnId
 from reqstool.common.utils import Utils
@@ -233,56 +234,13 @@ def __parse_location_git(self, locations_obj, instance_type, locations):
 
                 locations.append(git_location)
 
-    def __parse_requirement_filters(self, data) -> Dict[str, RequirementFilter]:  # NOSONAR
-        r_filters = {}
-
-        self.semantic_validator._validate_req_imports_filter_has_excludes_xor_includes(data)
-
-        if "filters" in data:
-            for urn in data["filters"].keys():
-                urn_filter = data["filters"][urn]
-
-                req_urn_ids_imports: Optional[Set[UrnId]] = None
-                req_urn_ids_excludes: Optional[Set[UrnId]] = None
-                custom_includes = None
-                custom_exclude = None
-
-                if "requirement_ids" in urn_filter:
-                    if "includes" in urn_filter["requirement_ids"]:
-                        filtered_ids = Utils.check_ids_to_filter(
-                            current_urn=urn, ids=urn_filter["requirement_ids"]["includes"]
-                        )
-                        req_ids_includes = set(filtered_ids)
-                        req_urn_ids_imports: Set[UrnId] = set(
-                            Utils.convert_ids_to_urn_id(urn=urn, ids=req_ids_includes)
-                        )
-
-                    if "excludes" in urn_filter["requirement_ids"]:
-                        filtered_ids = Utils.check_ids_to_filter(
-                            current_urn=urn, ids=urn_filter["requirement_ids"]["excludes"]
-                        )
-                        req_ids_excludes = set(filtered_ids)
-                        req_urn_ids_excludes: Set[UrnId] = set(
-                            Utils.convert_ids_to_urn_id(urn=urn, ids=req_ids_excludes)
-                        )
-
-                if "custom" in urn_filter:
-                    if "includes" in urn_filter["custom"]:
-                        custom_includes = urn_filter["custom"]["includes"]
-
-                    if "excludes" in urn_filter["custom"]:
-                        custom_exclude = urn_filter["custom"]["excludes"]
-
-                req_filter = RequirementFilter(
-                    urn_ids_imports=req_urn_ids_imports,
-                    urn_ids_excludes=req_urn_ids_excludes,
-                    custom_imports=custom_includes,
-                    custom_exclude=custom_exclude,
-                )
-
-                r_filters[urn] = req_filter
-
-        return r_filters
+    def __parse_requirement_filters(self, data) -> Dict[str, RequirementFilter]:
+        return parse_filters(
+            data=data,
+            ids_key="requirement_ids",
+            filter_cls=RequirementFilter,
+            validate_fn=self.semantic_validator._validate_req_imports_filter_has_excludes_xor_includes,
+        )
 
     @Requirements("REQ_004", "REQ_036")
     def __parse_requirements(self, model, data):  # NOSONAR

src/reqstool/model_generators/svcs_model_generator.py

@@ -1,11 +1,12 @@
 # Copyright © LFV
 
 import sys
-from typing import Dict, Optional, Set
+from typing import Dict
 
 from ruamel.yaml import YAML
 
 from reqstool.commands.exit_codes import EXIT_CODE_SYNTAX_VALIDATION_ERROR
+from reqstool.common.filter_parser import parse_filters
 from reqstool.common.models.lifecycle import LifecycleData
 from reqstool.common.models.urn_id import UrnId
 from reqstool.common.utils import Utils
@@ -71,47 +72,10 @@ def __parse_svcs(self, validated: SVCsPydanticModel) -> dict[UrnId, SVCData]:
 
         return r_result
 
-    def __parse_svc_filters(self, data) -> Dict[str, SVCFilter]:  # NOSONAR
-        r_filters = {}
-
-        self.semantic_validator._validate_svc_imports_filter_has_excludes_xor_includes(data)
-
-        if "filters" in data:
-            for urn in data["filters"].keys():
-                urn_filter = data["filters"][urn]
-
-                svc_urn_ids_includes: Optional[Set[UrnId]] = None
-                svc_urn_ids_excludes: Optional[Set[UrnId]] = None
-                custom_includes = None
-                custom_exclude = None
-
-                if "svc_ids" in urn_filter:
-                    if "includes" in urn_filter["svc_ids"]:
-                        svc_ids_includes = set(
-                            Utils.check_ids_to_filter(current_urn=urn, ids=urn_filter["svc_ids"]["includes"])
-                        )
-                        svc_urn_ids_includes = set(Utils.convert_ids_to_urn_id(urn=urn, ids=svc_ids_includes))
-
-                    if "excludes" in urn_filter["svc_ids"]:
-                        svc_ids_excludes = set(
-                            Utils.check_ids_to_filter(current_urn=urn, ids=urn_filter["svc_ids"]["excludes"])
-                        )
-                        svc_urn_ids_excludes = set(Utils.convert_ids_to_urn_id(urn=urn, ids=svc_ids_excludes))
-
-                if "custom" in urn_filter:
-                    if "includes" in urn_filter["custom"]:
-                        custom_includes = urn_filter["custom"]["includes"]
-
-                    if "excludes" in urn_filter["custom"]:
-                        custom_exclude = urn_filter["custom"]["excludes"]
-
-                svc_filter = SVCFilter(
-                    urn_ids_imports=svc_urn_ids_includes,
-                    urn_ids_excludes=svc_urn_ids_excludes,
-                    custom_imports=custom_includes,
-                    custom_exclude=custom_exclude,
-                )
-
-                r_filters[urn] = svc_filter
-
-        return r_filters
+    def __parse_svc_filters(self, data) -> Dict[str, SVCFilter]:
+        return parse_filters(
+            data=data,
+            ids_key="svc_ids",
+            filter_cls=SVCFilter,
+            validate_fn=self.semantic_validator._validate_svc_imports_filter_has_excludes_xor_includes,
+        )