lint fix

Author: Karthikeyannhs

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -22,16 +22,22 @@
     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]]
 
 
 @service
 class EligibilityCalculatorFactory:
     @staticmethod
-    def get(person_data: Row, campaign_configs: Collection[rules.CampaignConfig]) -> EligibilityCalculator:
-        return EligibilityCalculator(person_data=person_data, campaign_configs=campaign_configs)
+    def get(
+        person_data: Row, campaign_configs: Collection[rules.CampaignConfig]
+    ) -> EligibilityCalculator:
+        return EligibilityCalculator(
+            person_data=person_data, campaign_configs=campaign_configs
+        )
 
 
 @dataclass
@@ -51,34 +57,51 @@ 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()
         )
-        return set(cohorts_row.get("COHORT_MAP", {}).get("cohorts", {}).get("M", {}).keys())
 
     @staticmethod
-    def get_the_best_cohort_memberships(cohort_results: dict[str, CohortResult]) -> tuple[Status, list[CohortResult]]:
+    def get_the_best_cohort_memberships(
+        cohort_results: dict[str, CohortResult],
+    ) -> tuple[Status, list[CohortResult]]:
         """
         1. Get all the cohorts with the best status
         2. Case 1: Ignore magic cohort if other cohorts have better status
-           Case 2: If no cohorts have the better status than magic cohort,
-                    then response excludes cohort memberships but will have actions/suitability rules
-                    also, excludes the cohorts with no positive or negative description
+           Case 2: If no cohorts have the better status than magic cohort, then response excludes cohort memberships but
+           will have actions/suitability rules also excludes the cohorts with no positive or negative description
         """
         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]
-        if all(cc.cohort_code and str(cc.cohort_code.upper()) == MAGIC_COHORT_LABEL.upper() for cc in best_cohorts):
+        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
+        ]
+        if all(
+            cc.cohort_code and str(cc.cohort_code.upper()) == MAGIC_COHORT_LABEL.upper()
+            for cc in best_cohorts
+        ):
             # Update the magic cohort to have no cohort membership information
             best_cohorts = [
-                CohortResult(cohort_code="", status=best_status, reasons=best_cohorts[0].reasons, description="")
+                CohortResult(
+                    cohort_code="",
+                    status=best_status,
+                    reasons=best_cohorts[0].reasons,
+                    description="",
+                )
             ]
         else:
             best_cohorts = [
@@ -99,15 +122,22 @@ def get_exclusion_rules(
             for ir in filter_rules
             if ir.cohort_label is None
             or cohort.cohort_label == ir.cohort_label
-            or (isinstance(ir.cohort_label, (list, set, tuple)) and cohort.cohort_label in ir.cohort_label)
+            or (
+                isinstance(ir.cohort_label, (list, set, tuple))
+                and cohort.cohort_label in ir.cohort_label
+            )
         )
 
     @staticmethod
     def get_rules_by_type(
         active_iteration: Iteration,
     ) -> tuple[tuple[rules.IterationRule, ...], tuple[rules.IterationRule, ...]]:
         filter_rules, suppression_rules = (
-            tuple(rule for rule in active_iteration.iteration_rules if attrgetter("type")(rule) == rule_type)
+            tuple(
+                rule
+                for rule in active_iteration.iteration_rules
+                if attrgetter("type")(rule) == rule_type
+            )
             for rule_type in (rules.RuleType.filter, rules.RuleType.suppression)
         )
         return filter_rules, suppression_rules
@@ -122,33 +152,54 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             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)
+                filter_rules, suppression_rules = self.get_rules_by_type(
+                    active_iteration
+                )
 
-                for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
+                for cohort in sorted(
+                    active_iteration.iteration_cohorts, key=attrgetter("priority")
+                ):
                     # Base Eligibility - check
-                    if cohort.cohort_label in self.person_cohorts or active_iteration.has_magic_cohort:
+                    if (
+                        cohort.cohort_label in self.person_cohorts
+                        or active_iteration.has_magic_cohort
+                    ):
                         # Eligibility - check
-                        if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
+                        if self.is_eligible_by_filter_rules(
+                            cohort, cohort_results, filter_rules
+                        ):
                             # Actionability - evaluation
-                            self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
+                            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,
+                            (
+                                cohort.cohort_group
+                                if cohort.cohort_group
+                                else cohort.cohort_label
+                            ),
                             Status.not_eligible,
                             [],
                             str(cohort.negative_description),
                         )
 
                 # Determine Result between cohorts - get the best
-                status, best_cohorts = self.get_the_best_cohort_memberships(cohort_results)
+                status, best_cohorts = self.get_the_best_cohort_memberships(
+                    cohort_results
+                )
 
-                iteration_results[active_iteration.name] = IterationResult(status, best_cohorts)
+                iteration_results[active_iteration.name] = IterationResult(
+                    status, best_cohorts
+                )
 
             # Determine results between iterations - get the best
             if iteration_results:
-                best_candidate = max(iteration_results.values(), key=lambda r: r.status.value)
+                best_candidate = max(
+                    iteration_results.values(), key=lambda r: r.status.value
+                )
             else:
                 best_candidate = IterationResult(eligibility.Status.not_eligible, [])
             condition_results[condition_name] = best_candidate
@@ -158,19 +209,25 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
         return eligibility.EligibilityStatus(conditions=final_result)
 
     @staticmethod
-    def build_condition_results(condition_results: dict[ConditionName, IterationResult]) -> list[Condition]:
+    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)
+                    grouped_cohort_results[cohort_result.cohort_code].append(
+                        cohort_result
+                    )
 
             # deduplicate grouped cohort results by cohort_code
             deduplicated_cohort_results = {
-                cohort_code: results[0] for cohort_code, results in grouped_cohort_results.items() if results
+                cohort_code: results[0]
+                for cohort_code, results in grouped_cohort_results.items()
+                if results
             }
 
             # return condition with cohort results
@@ -191,16 +248,22 @@ def is_eligible_by_filter_rules(
     ) -> bool:
         is_eligible = True
         priority_getter = attrgetter("priority")
-        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, filter_rules), key=priority_getter)
+        sorted_rules_by_priority = sorted(
+            self.get_exclusion_rules(cohort, filter_rules), key=priority_getter
+        )
 
         for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
-            status, group_inclusion_reasons, group_exclusion_reasons, rule_stop = self.evaluate_rules_priority_group(
-                rule_group
+            status, group_inclusion_reasons, group_exclusion_reasons, rule_stop = (
+                self.evaluate_rules_priority_group(rule_group)
             )
             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.cohort_group
+                            if cohort.cohort_group
+                            else cohort.cohort_label
+                        ),
                         Status.not_eligible,
                         [],
                         str(cohort.negative_description),
@@ -219,11 +282,13 @@ def evaluate_suppression_rules(
         priority_getter = attrgetter("priority")
         suppression_reasons = []
 
-        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, suppression_rules), key=priority_getter)
+        sorted_rules_by_priority = sorted(
+            self.get_exclusion_rules(cohort, suppression_rules), key=priority_getter
+        )
 
         for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
-            status, group_inclusion_reasons, group_exclusion_reasons, rule_stop = self.evaluate_rules_priority_group(
-                rule_group
+            status, group_inclusion_reasons, group_exclusion_reasons, rule_stop = (
+                self.evaluate_rules_priority_group(rule_group)
             )
             if status.is_exclusion:
                 is_actionable = False
@@ -250,7 +315,9 @@ def evaluate_suppression_rules(
 
     def evaluate_rules_priority_group(
         self, rules_group: Iterator[rules.IterationRule]
-    ) -> tuple[eligibility.Status, list[eligibility.Reason], list[eligibility.Reason], bool]:
+    ) -> tuple[
+        eligibility.Status, list[eligibility.Reason], list[eligibility.Reason], bool
+    ]:
         is_rule_stop = False
         inclusion_reasons, exclusion_reasons = [], []
         best_status = eligibility.Status.not_eligible