Merge remote-tracking branch 'origin/main' into feature/eja-eli-279-tighten-up-gitlab-permissions

Author: eddalmond1

poetry.lock

@@ -24,7 +24,7 @@ python = "^3.13"
 flask = {extras = ["async"], version = "^3.1.1"}
 httpx = "^0.28.1"
 yarl = "^1.18.3"
-pydantic = "^2.10.6"
+pydantic = "^2.11.7"
 asgiref = "^3.8.1"
 boto3 = "^1.37.3"
 botocore = "^1.38.32"

pyproject.toml

@@ -8,7 +8,7 @@
 from mangum.types import LambdaContext, LambdaEvent
 
 from eligibility_signposting_api import repos, services
-from eligibility_signposting_api.config import config, init_logging
+from eligibility_signposting_api.config.config import config, init_logging
 from eligibility_signposting_api.error_handler import handle_exception
 from eligibility_signposting_api.views import eligibility_blueprint
 

src/eligibility_signposting_api/app.py

@@ -0,0 +1 @@
+MAGIC_COHORT_LABEL = "elid_all_people"

src/eligibility_signposting_api/config/__init__.py

@@ -12,7 +12,7 @@
 ConditionName = NewType("ConditionName", str)
 
 RuleName = NewType("RuleName", str)
-RuleResult = NewType("RuleResult", str)
+RuleDescription = NewType("RuleDescription", str)
 
 
 class RuleType(StrEnum):
@@ -60,28 +60,28 @@ def best(*statuses: Status) -> Status:
 class Reason:
     rule_type: RuleType
     rule_name: RuleName
-    rule_result: RuleResult
+    rule_description: RuleDescription | None
 
 
 @dataclass
 class Condition:
     condition_name: ConditionName
     status: Status
-    cohort_results: list[CohortResult]
+    cohort_results: list[CohortGroupResult]
 
 
 @dataclass
-class CohortResult:
+class CohortGroupResult:
     cohort_code: str
     status: Status
     reasons: list[Reason]
-    description: str
+    description: str | None
 
 
 @dataclass
 class IterationResult:
     status: Status
-    cohort_results: list[CohortResult]
+    cohort_results: list[CohortGroupResult]
 
 
 @dataclass

src/eligibility_signposting_api/config.py …ibility_signposting_api/config/config.pysrc/eligibility_signposting_api/config.py renamed to src/eligibility_signposting_api/config/config.py src/eligibility_signposting_api/config.py renamed to src/eligibility_signposting_api/config/config.py

@@ -10,6 +10,8 @@
 
 from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
 
+from eligibility_signposting_api.config.contants import MAGIC_COHORT_LABEL
+
 if typing.TYPE_CHECKING:  # pragma: no cover
     from pydantic import SerializationInfo
 
@@ -85,14 +87,18 @@ class RuleAttributeLevel(StrEnum):
 
 
 class IterationCohort(BaseModel):
-    cohort_label: CohortLabel | None = Field(None, alias="CohortLabel")
-    cohort_group: CohortGroup | None = Field(None, alias="CohortGroup")
+    cohort_label: CohortLabel = Field(alias="CohortLabel")
+    cohort_group: CohortGroup = Field(alias="CohortGroup")
     positive_description: Description | None = Field(None, alias="PositiveDescription")
     negative_description: Description | None = Field(None, alias="NegativeDescription")
     priority: int | None = Field(None, alias="Priority")
 
     model_config = {"populate_by_name": True, "extra": "ignore"}
 
+    @cached_property
+    def is_magic_cohort(self) -> bool:
+        return self.cohort_label.upper() == MAGIC_COHORT_LABEL.upper()
+
 
 class IterationRule(BaseModel):
     type: RuleType = Field(..., alias="Type")
@@ -107,14 +113,14 @@ class IterationRule(BaseModel):
     attribute_target: RuleAttributeTarget | None = Field(None, alias="AttributeTarget")
     rule_stop: RuleStop = Field(RuleStop(False), alias="RuleStop")  # noqa: FBT003
 
+    model_config = {"populate_by_name": True, "extra": "ignore"}
+
     @field_validator("rule_stop", mode="before")
     def parse_yn_to_bool(cls, v: str | bool) -> bool:  # noqa: N805
         if isinstance(v, str):
             return v.upper() == "Y"
         return v
 
-    model_config = {"populate_by_name": True, "extra": "ignore"}
-
 
 class Iteration(BaseModel):
     id: IterationID = Field(..., alias="ID")

src/eligibility_signposting_api/config/contants.py

@@ -7,7 +7,7 @@
 from wireup import Inject, service
 from yarl import URL
 
-from eligibility_signposting_api.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
+from eligibility_signposting_api.config.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
 
 logger = logging.getLogger(__name__)
 

src/eligibility_signposting_api/model/eligibility.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from _operator import attrgetter
+from collections import defaultdict
 from collections.abc import Collection, Iterable, Iterator, Mapping
 from dataclasses import dataclass, field
 from itertools import groupby
@@ -13,16 +14,17 @@
 
 from eligibility_signposting_api.model import eligibility, rules
 from eligibility_signposting_api.model.eligibility import (
-    CohortResult,
+    CohortGroupResult,
     Condition,
     ConditionName,
     IterationResult,
     Status,
 )
-from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
+from eligibility_signposting_api.services.calculators.rule_calculator import (
+    RuleCalculator,
+)
 
 Row = Collection[Mapping[str, Any]]
-magic_cohort = "elid_all_people"
 
 
 @service
@@ -49,23 +51,39 @@ def campaigns_grouped_by_condition_name(
     ) -> Iterator[tuple[eligibility.ConditionName, list[rules.CampaignConfig]]]:
         """Generator function to iterate over campaign groups by condition name."""
         for condition_name, campaign_group in groupby(
-            sorted(self.active_campaigns, key=attrgetter("target")), key=attrgetter("target")
+            sorted(self.active_campaigns, key=attrgetter("target")),
+            key=attrgetter("target"),
         ):
             yield condition_name, list(campaign_group)
 
     @property
     def person_cohorts(self) -> set[str]:
         cohorts_row: Mapping[str, dict[str, dict[str, dict[str, Any]]]] = next(
-            (row for row in self.person_data if row.get("ATTRIBUTE_TYPE") == "COHORTS"), {}
+            (row for row in self.person_data if row.get("ATTRIBUTE_TYPE") == "COHORTS"),
+            {},
         )
         return set(cohorts_row.get("COHORT_MAP", {}).get("cohorts", {}).get("M", {}).keys())
 
     @staticmethod
-    def get_best_cohort(cohort_results: dict[str, CohortResult]) -> tuple[Status, list[CohortResult]]:
+    def get_the_best_cohort_memberships(
+        cohort_results: dict[str, CohortGroupResult],
+    ) -> tuple[Status, list[CohortGroupResult]]:
         if not cohort_results:
             return eligibility.Status.not_eligible, []
+
         best_status = eligibility.Status.best(*[result.status for result in cohort_results.values()])
         best_cohorts = [result for result in cohort_results.values() if result.status == best_status]
+
+        best_cohorts = [
+            CohortGroupResult(
+                cohort_code=cc.cohort_code,
+                status=cc.status,
+                reasons=cc.reasons,
+                description=(cc.description or "").strip() if cc.description else "",
+            )
+            for cc in best_cohorts
+        ]
+
         return best_status, best_cohorts
 
     @staticmethod
@@ -98,28 +116,11 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             iteration_results: dict[str, IterationResult] = {}
 
             for active_iteration in [cc.current_iteration for cc in campaign_group]:
-                cohort_results: dict[str, CohortResult] = {}
-
-                filter_rules, suppression_rules = self.get_rules_by_type(active_iteration)
-                for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
-                    # Base Eligibility - check
-                    if cohort.cohort_label in self.person_cohorts or cohort.cohort_label == magic_cohort:
-                        # Eligibility - check
-                        if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
-                            # Actionability - evaluation
-                            self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
-
-                    # Not base eligible
-                    elif cohort.cohort_label is not None:
-                        cohort_results[cohort.cohort_label] = CohortResult(
-                            cohort.cohort_group if cohort.cohort_group else cohort.cohort_label,
-                            Status.not_eligible,
-                            [],
-                            str(cohort.negative_description),
-                        )
+                cohort_results: dict[str, CohortGroupResult] = self.get_cohort_results(active_iteration)
 
                 # Determine Result between cohorts - get the best
-                status, best_cohorts = self.get_best_cohort(cohort_results)
+                status, best_cohorts = self.get_the_best_cohort_memberships(cohort_results)
+
                 iteration_results[active_iteration.name] = IterationResult(status, best_cohorts)
 
             # Determine results between iterations - get the best
@@ -130,20 +131,71 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             condition_results[condition_name] = best_candidate
 
         # Consolidate all the results and return
-        final_result = [
-            Condition(
-                condition_name=condition_name,
-                status=active_iteration_result.status,
-                cohort_results=active_iteration_result.cohort_results,
-            )
-            for condition_name, active_iteration_result in condition_results.items()
-        ]
+        final_result = self.build_condition_results(condition_results)
         return eligibility.EligibilityStatus(conditions=final_result)
 
+    def get_cohort_results(self, active_iteration: rules.Iteration) -> dict[str, CohortGroupResult]:
+        cohort_results: dict[str, CohortGroupResult] = {}
+        filter_rules, suppression_rules = self.get_rules_by_type(active_iteration)
+        for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
+            # Base Eligibility - check
+            if cohort.cohort_label in self.person_cohorts or cohort.is_magic_cohort:
+                # Eligibility - check
+                if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
+                    # Actionability - evaluation
+                    self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
+
+            # Not base eligible
+            elif cohort.cohort_label is not None:
+                cohort_results[cohort.cohort_label] = CohortGroupResult(
+                    (cohort.cohort_group),
+                    Status.not_eligible,
+                    [],
+                    cohort.negative_description,
+                )
+        return cohort_results
+
+    @staticmethod
+    def build_condition_results(
+        condition_results: dict[ConditionName, IterationResult],
+    ) -> list[Condition]:
+        conditions: list[Condition] = []
+        # iterate over conditions
+        for condition_name, active_iteration_result in condition_results.items():
+            grouped_cohort_results = defaultdict(list)
+            # iterate over cohorts and group them by status and cohort_group
+            for cohort_result in active_iteration_result.cohort_results:
+                if active_iteration_result.status == cohort_result.status:
+                    grouped_cohort_results[cohort_result.cohort_code].append(cohort_result)
+
+            # deduplicate grouped cohort results by cohort_code
+            deduplicated_cohort_results = [
+                CohortGroupResult(
+                    cohort_code=group_cohort_code,
+                    status=group[0].status,
+                    # Flatten all reasons from the group
+                    reasons=[reason for cohort in group for reason in cohort.reasons],
+                    # get the first nonempty description
+                    description=next((c.description for c in group if c.description), group[0].description),
+                )
+                for group_cohort_code, group in grouped_cohort_results.items()
+                if group
+            ]
+
+            # return condition with cohort results
+            conditions.append(
+                Condition(
+                    condition_name=condition_name,
+                    status=active_iteration_result.status,
+                    cohort_results=list(deduplicated_cohort_results),
+                )
+            )
+        return conditions
+
     def is_eligible_by_filter_rules(
         self,
         cohort: IterationCohort,
-        cohort_results: dict[str, CohortResult],
+        cohort_results: dict[str, CohortGroupResult],
         filter_rules: Iterable[rules.IterationRule],
     ) -> bool:
         is_eligible = True
@@ -156,11 +208,11 @@ def is_eligible_by_filter_rules(
             )
             if status.is_exclusion:
                 if cohort.cohort_label is not None:
-                    cohort_results[str(cohort.cohort_label)] = CohortResult(
-                        cohort.cohort_group if cohort.cohort_group else cohort.cohort_label,
+                    cohort_results[cohort.cohort_label] = CohortGroupResult(
+                        (cohort.cohort_group),
                         Status.not_eligible,
                         [],
-                        str(cohort.negative_description),
+                        cohort.negative_description,
                     )
                 is_eligible = False
                 break
@@ -169,7 +221,7 @@ def is_eligible_by_filter_rules(
     def evaluate_suppression_rules(
         self,
         cohort: IterationCohort,
-        cohort_results: dict[str, CohortResult],
+        cohort_results: dict[str, CohortGroupResult],
         suppression_rules: Iterable[rules.IterationRule],
     ) -> None:
         is_actionable: bool = True
@@ -191,18 +243,18 @@ def evaluate_suppression_rules(
         if cohort.cohort_label is not None:
             key = cohort.cohort_label
             if is_actionable:
-                cohort_results[key] = CohortResult(
-                    cohort.cohort_group if cohort.cohort_group else key,
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group,
                     Status.actionable,
                     [],
-                    str(cohort.positive_description),
+                    cohort.positive_description,
                 )
             else:
-                cohort_results[key] = CohortResult(
-                    cohort.cohort_group if cohort.cohort_group else key,
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group,
                     Status.not_actionable,
                     suppression_reasons,
-                    str(cohort.positive_description),
+                    cohort.positive_description,
                 )
 
     def evaluate_rules_priority_group(