ELI-351: Extracts Person data class

Author: shweta-nhs

src/eligibility_signposting_api/model/person.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class Person:
+    data: list[dict[str, Any]]

src/eligibility_signposting_api/repos/person_repo.py

@@ -6,6 +6,7 @@
 from wireup import Inject, service
 
 from eligibility_signposting_api.model.eligibility_status import NHSNumber
+from eligibility_signposting_api.model.person import Person
 from eligibility_signposting_api.repos.exceptions import NotFoundError
 
 logger = logging.getLogger(__name__)
@@ -35,7 +36,7 @@ def __init__(self, table: Annotated[Any, Inject(qualifier="person_table")]) -> N
         super().__init__()
         self.table = table
 
-    def get_eligibility_data(self, nhs_number: NHSNumber) -> list[dict[str, Any]]:
+    def get_eligibility_data(self, nhs_number: NHSNumber) -> Person:
         response = self.table.query(KeyConditionExpression=Key("NHS_NUMBER").eq(nhs_number))
         logger.debug("response %r for %r", response, nhs_number, extra={"response": response, "nhs_number": nhs_number})
 
@@ -44,4 +45,5 @@ def get_eligibility_data(self, nhs_number: NHSNumber) -> list[dict[str, Any]]:
             raise NotFoundError(message)
 
         logger.debug("returning items %s", items, extra={"items": items})
-        return items
+
+        return Person(data=items)

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -2,10 +2,9 @@
 
 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
-from typing import Any
+from typing import TYPE_CHECKING
 
 from wireup import service
 
@@ -43,19 +42,22 @@
 from eligibility_signposting_api.services.processors.campaign_evaluator import CampaignEvaluator
 from eligibility_signposting_api.services.processors.person_data_reader import PersonDataReader
 
-Row = Collection[Mapping[str, Any]]
+if TYPE_CHECKING:
+    from collections.abc import Collection, Iterable, Iterator
+
+    from eligibility_signposting_api.model.person import Person
 
 
 @service
 class EligibilityCalculatorFactory:
     @staticmethod
-    def get(person_data: Row, campaign_configs: Collection[CampaignConfig]) -> EligibilityCalculator:
-        return EligibilityCalculator(person_data=person_data, campaign_configs=campaign_configs)
+    def get(person: Person, campaign_configs: Collection[CampaignConfig]) -> EligibilityCalculator:
+        return EligibilityCalculator(person=person, campaign_configs=campaign_configs)
 
 
 @dataclass
 class EligibilityCalculator:
-    person_data: Row
+    person: Person
     campaign_configs: Collection[CampaignConfig]
 
     campaign_evaluator: CampaignEvaluator = field(default_factory=CampaignEvaluator)
@@ -238,7 +240,7 @@ def handle_action_rules(
         for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
             rule_group_list = list(rule_group)
             matcher_matched_list = [
-                RuleCalculator(person_data=self.person_data, rule=rule).evaluate_exclusion()[1].matcher_matched
+                RuleCalculator(person=self.person, rule=rule).evaluate_exclusion()[1].matcher_matched
                 for rule in rule_group_list
             ]
 
@@ -258,7 +260,7 @@ def get_cohort_results(self, active_iteration: Iteration) -> dict[str, CohortGro
         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
-            person_cohorts = self.person_data_reader.get_person_cohorts(self.person_data)
+            person_cohorts = self.person_data_reader.get_person_cohorts(self.person)
             if cohort.cohort_label in person_cohorts or cohort.is_magic_cohort:
                 # Eligibility - check
                 if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
@@ -329,7 +331,7 @@ def is_eligible_by_filter_rules(
             if status.is_exclusion:
                 if cohort.cohort_label is not None:
                     cohort_results[cohort.cohort_label] = CohortGroupResult(
-                        (cohort.cohort_group),
+                        cohort.cohort_group,
                         Status.not_eligible,
                         [],
                         cohort.negative_description,
@@ -383,7 +385,7 @@ def evaluate_rules_priority_group(
 
         for rule in rules_group:
             is_rule_stop = rule.rule_stop or is_rule_stop
-            rule_calculator = RuleCalculator(person_data=self.person_data, rule=rule)
+            rule_calculator = RuleCalculator(person=self.person, rule=rule)
             status, reason = rule_calculator.evaluate_exclusion()
             if status.is_exclusion:
                 best_status = eligibility_status.Status.best(status, best_status)

src/eligibility_signposting_api/services/calculators/rule_calculator.py

@@ -1,8 +1,7 @@
 from __future__ import annotations
 
-from collections.abc import Collection, Mapping
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING
 
 from hamcrest.core.string_description import StringDescription
 
@@ -11,12 +10,15 @@
 from eligibility_signposting_api.services.operators.operators import OperatorRegistry
 from eligibility_signposting_api.services.processors.person_data_reader import PersonDataReader
 
-Row = Collection[Mapping[str, Any]]
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+    from eligibility_signposting_api.model.person import Person
 
 
 @dataclass
 class RuleCalculator:
-    person_data: Row
+    person: Person
     rule: IterationRule
 
     person_data_reader: PersonDataReader = field(default_factory=PersonDataReader)
@@ -39,22 +41,22 @@ def get_attribute_value(self) -> str | None:
         match self.rule.attribute_level:
             case RuleAttributeLevel.PERSON:
                 person: Mapping[str, str | None] | None = next(
-                    (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == "PERSON"), None
+                    (r for r in self.person.data if r.get("ATTRIBUTE_TYPE", "") == "PERSON"), None
                 )
                 attribute_value = person.get(str(self.rule.attribute_name)) if person else None
             case RuleAttributeLevel.COHORT:
                 cohorts: Mapping[str, str | None] | None = next(
-                    (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == "COHORTS"), None
+                    (r for r in self.person.data if r.get("ATTRIBUTE_TYPE", "") == "COHORTS"), None
                 )
                 if cohorts:
-                    person_cohorts = self.person_data_reader.get_person_cohorts(self.person_data)
+                    person_cohorts = self.person_data_reader.get_person_cohorts(self.person)
                     attribute_value = ",".join(person_cohorts)
                 else:
                     attribute_value = None
 
             case RuleAttributeLevel.TARGET:
                 target: Mapping[str, str | None] | None = next(
-                    (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == self.rule.attribute_target), None
+                    (r for r in self.person.data if r.get("ATTRIBUTE_TYPE", "") == self.rule.attribute_target), None
                 )
                 attribute_value = target.get(str(self.rule.attribute_name)) if target else None
             case _:  # pragma: no cover

src/eligibility_signposting_api/services/processors/person_data_reader.py

@@ -1,26 +1,25 @@
 from __future__ import annotations
 
-from collections.abc import Collection, Mapping
-from typing import Any
-
 from wireup import service
 
-Row = Collection[Mapping[str, Any]]
+from eligibility_signposting_api.model.person import Person
 
 
 @service
 class PersonDataReader:
     """Handles extracting and interpreting person data."""
 
-    def get_person_cohorts(self, person_data: Row) -> set[str]:
-        cohorts_row: Mapping[str, list[dict[str, str]]] = next(
-            (row for row in person_data if row.get("ATTRIBUTE_TYPE") == "COHORTS"),
-            {},
-        )
+    def get_person_cohorts(self, person: Person) -> set[str]:
+        cohorts_row: Person = Person([])
+        for data in person.data:
+            if data.get("ATTRIBUTE_TYPE") == "COHORTS":
+                cohorts_row.data.append(data)
+
         person_cohorts = set()
 
-        for membership in cohorts_row.get("COHORT_MEMBERSHIPS", []):
-            if membership.get("COHORT_LABEL"):
-                person_cohorts.add(membership.get("COHORT_LABEL"))
+        if cohorts_row.data:
+            for membership in cohorts_row.data[0].get("COHORT_MEMBERSHIPS", []):
+                if membership.get("COHORT_LABEL"):
+                    person_cohorts.add(membership.get("COHORT_LABEL"))
 
         return person_cohorts

tests/fixtures/builders/repos/person.py

@@ -5,6 +5,7 @@
 
 from faker import Faker
 
+from eligibility_signposting_api.model.person import Person
 from tests.conftest import PersonDetailProvider
 
 Gender = Literal["0", "1", "2", "9"]  # 0 - Not known, 1- Male, 2 - Female, 9 - Not specified. I know, right?
@@ -27,7 +28,7 @@ def person_rows_builder(  # noqa:PLR0913
     de: bool | None = ...,
     msoa: str | None = ...,
     lsoa: str | None = ...,
-) -> list[dict[str, Any]]:
+) -> Person:
     faker = Faker("en_UK")
     faker.add_provider(PersonDetailProvider)
 
@@ -85,4 +86,5 @@ def person_rows_builder(  # noqa:PLR0913
     )
 
     shuffle(rows)
-    return rows
+
+    return Person(data=rows)

tests/integration/conftest.py

@@ -336,7 +336,7 @@ def persisted_person(person_table: Any, faker: Faker) -> Generator[eligibility_s
     date_of_birth = eligibility_status.DateOfBirth(faker.date_of_birth(minimum_age=18, maximum_age=65))
 
     for row in (
-        rows := person_rows_builder(nhs_number, date_of_birth=date_of_birth, postcode="hp1", cohorts=["cohort1"])
+        rows := person_rows_builder(nhs_number, date_of_birth=date_of_birth, postcode="hp1", cohorts=["cohort1"]).data
     ):
         person_table.put_item(Item=row)
 
@@ -357,7 +357,7 @@ def persisted_77yo_person(person_table: Any, faker: Faker) -> Generator[eligibil
             date_of_birth=date_of_birth,
             postcode="hp1",
             cohorts=["cohort1", "cohort2"],
-        )
+        ).data
     ):
         person_table.put_item(Item=row)
 
@@ -379,7 +379,7 @@ def persisted_person_all_cohorts(person_table: Any, faker: Faker) -> Generator[e
             postcode="hp1",
             cohorts=["cohort_label1", "cohort_label2", "cohort_label3"],
             icb="QE1",
-        )
+        ).data
     ):
         person_table.put_item(Item=row)
 
@@ -393,7 +393,7 @@ def persisted_person_all_cohorts(person_table: Any, faker: Faker) -> Generator[e
 def persisted_person_no_cohorts(person_table: Any, faker: Faker) -> Generator[eligibility_status.NHSNumber]:
     nhs_number = eligibility_status.NHSNumber(faker.nhs_number())
 
-    for row in (rows := person_rows_builder(nhs_number)):
+    for row in (rows := person_rows_builder(nhs_number).data):
         person_table.put_item(Item=row)
 
     yield nhs_number
@@ -407,7 +407,7 @@ def persisted_person_pc_sw19(person_table: Any, faker: Faker) -> Generator[eligi
     nhs_number = eligibility_status.NHSNumber(
         faker.nhs_number(),
     )
-    for row in (rows := person_rows_builder(nhs_number, postcode="SW19", cohorts=["cohort1"])):
+    for row in (rows := person_rows_builder(nhs_number, postcode="SW19", cohorts=["cohort1"]).data):
         person_table.put_item(Item=row)
 
     yield nhs_number

tests/integration/repo/test_person_repo.py

@@ -18,7 +18,7 @@ def test_person_found(person_table: Any, persisted_person: NHSNumber):
 
     # Then
     assert_that(
-        actual,
+        actual.data,
         contains_inanyorder(
             has_entries({"NHS_NUMBER": persisted_person, "ATTRIBUTE_TYPE": "PERSON"}),
             has_entries({"NHS_NUMBER": persisted_person, "ATTRIBUTE_TYPE": "COHORTS"}),

tests/unit/services/calculators/test_eligibility_calculator.py

@@ -40,6 +40,7 @@
     UrlLabel,
     UrlLink,
 )
+from eligibility_signposting_api.model.person import Person
 from eligibility_signposting_api.services.calculators.eligibility_calculator import EligibilityCalculator
 from tests.fixtures.builders.model import rule as rule_builder
 from tests.fixtures.builders.repos.person import person_rows_builder
@@ -956,12 +957,12 @@ def test_status_on_cohort_attribute_level(
     # Given
     nhs_number = NHSNumber(faker.nhs_number())
 
-    person_row: list[dict[str, Any]] = person_rows_builder(
-        nhs_number, cohorts=["cohort1", "covid_eligibility_complaint_list"]
-    )
-    person_row_with_extra_items_in_cohort_row = [
-        {**r, "LOCATION": "HP1"} for r in person_row if r.get("ATTRIBUTE_TYPE", "") == "COHORTS"
-    ]
+    person_row: Person = person_rows_builder(nhs_number, cohorts=["cohort1", "covid_eligibility_complaint_list"])
+
+    person_row_with_extra_items_in_cohort_row = Person(person_row.data)
+    for row in person_row_with_extra_items_in_cohort_row.data:
+        if row.get("ATTRIBUTE_TYPE", "") == "COHORTS":
+            row["LOCATION"] = "HP1"
 
     campaign_configs = [
         rule_builder.CampaignConfigFactory.build(

tests/unit/services/calculators/test_rule_calculator.py

@@ -1,29 +1,25 @@
-from collections.abc import Collection, Mapping
-from typing import Any
-
 import pytest
 
 from eligibility_signposting_api.model.campaign_config import IterationRule, RuleAttributeLevel
+from eligibility_signposting_api.model.person import Person
 from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
 from tests.fixtures.builders.model import rule as rule_builder
 
-Row = Collection[Mapping[str, Any]]
-
 
 @pytest.mark.parametrize(
     ("person_data", "rule", "expected"),
     [
         # PERSON attribute level
         (
-            [{"ATTRIBUTE_TYPE": "PERSON", "POSTCODE": "SW19"}],
+            Person([{"ATTRIBUTE_TYPE": "PERSON", "POSTCODE": "SW19"}]),
             rule_builder.IterationRuleFactory.build(
                 attribute_level=RuleAttributeLevel.PERSON, attribute_name="POSTCODE"
             ),
             "SW19",
         ),
         # TARGET attribute level
         (
-            [{"ATTRIBUTE_TYPE": "RSV", "LAST_SUCCESSFUL_DATE": "20240101"}],
+            Person([{"ATTRIBUTE_TYPE": "RSV", "LAST_SUCCESSFUL_DATE": "20240101"}]),
             rule_builder.IterationRuleFactory.build(
                 attribute_level=RuleAttributeLevel.TARGET,
                 attribute_name="LAST_SUCCESSFUL_DATE",
@@ -33,17 +29,17 @@
         ),
         # COHORT attribute level
         (
-            [{"ATTRIBUTE_TYPE": "COHORTS", "COHORT_LABEL": ""}],
+            Person([{"ATTRIBUTE_TYPE": "COHORTS", "COHORT_LABEL": ""}]),
             rule_builder.IterationRuleFactory.build(
                 attribute_level=RuleAttributeLevel.COHORT, attribute_name="COHORT_LABEL"
             ),
             "",
         ),
     ],
 )
-def test_get_attribute_value_for_all_attribute_levels(person_data: Row, rule: IterationRule, expected: str):
+def test_get_attribute_value_for_all_attribute_levels(person_data: Person, rule: IterationRule, expected: str):
     # Given
-    calc = RuleCalculator(person_data=person_data, rule=rule)
+    calc = RuleCalculator(person=person_data, rule=rule)
     # When
     actual = calc.get_attribute_value()
     # Then