Merge pull request #43 from NHSDigital/feature/eli-21-test-simple-rule

ELI-210: Test simple rule

Author: acerathereinspirative-nhs-nvir

poetry.lock

@@ -33,6 +33,7 @@ mangum = "^0.19.0"
 wireup = "^1.0.1"
 python-json-logger = "^3.3.0"
 fhir-resources = "^8.0.0"
+python-dateutil = "^2.9.0"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.11.0"

pyproject.toml

@@ -24,4 +24,4 @@ def handle_exception(e: Exception) -> ResponseReturnValue | HTTPException:
             )  # pyright: ignore[reportCallIssue]
         ]
     )
-    return make_response(problem.model_dump(), HTTPStatus.INTERNAL_SERVER_ERROR)
+    return make_response(problem.model_dump(by_alias=True), HTTPStatus.INTERNAL_SERVER_ERROR)

src/eligibility_signposting_api/error_handler.py

@@ -8,5 +8,7 @@
 Postcode = NewType("Postcode", str)
 
 
-class Eligibility(BaseModel):
-    processed_suggestions: list[dict]
+class EligibilityStatus(BaseModel):
+    eligible: bool
+    reasons: list[dict]
+    actions: list[dict]

src/eligibility_signposting_api/model/eligibility.py

@@ -17,10 +17,10 @@
 IterationName = NewType("IterationName", str)
 IterationVersion = NewType("IterationVersion", str)
 IterationID = NewType("IterationID", str)
+IterationDate = NewType("IterationDate", date)
 RuleName = NewType("RuleName", str)
 RuleDescription = NewType("RuleDescription", str)
 RulePriority = NewType("RulePriority", int)
-RuleAttributeLevel = NewType("RuleAttributeLevel", str)
 RuleAttributeName = NewType("RuleAttributeName", str)
 RuleComparator = NewType("RuleComparator", str)
 StartDate = NewType("StartDate", date)
@@ -34,17 +34,24 @@ class RuleType(str, Enum):
 
 
 class RuleOperator(str, Enum):
+    equals = "="
+    ne = "!="
     lt = "<"
+    lte = "<="
     gt = ">"
+    gte = ">="
     year_gt = "Y>"
     not_in = "not_in"
-    equals = "="
-    lte = "<="
-    ne = "!="
     date_gte = "D>="
     member_of = "MemberOf"
 
 
+class RuleAttributeLevel(str, Enum):
+    PERSON = "PERSON"
+    TARGET = "TARGET"
+    COHORT = "COHORT"
+
+
 class IterationCohort(BaseModel):
     cohort_label: str | None = Field(None, alias="CohortLabel")
     priority: int | None = Field(None, alias="Priority")
@@ -62,26 +69,38 @@ class IterationRule(BaseModel):
     operator: RuleOperator = Field(..., alias="Operator")
     comparator: RuleComparator = Field(..., alias="Comparator")
     attribute_target: str | None = Field(None, alias="AttributeTarget")
-    comms_routing: str | None = Field(None, alias="CommsRouting")
 
     model_config = {"populate_by_name": True}
 
 
 class Iteration(BaseModel):
     id: IterationID = Field(..., alias="ID")
-    default_comms_routing: str | None = Field(None, alias="DefaultCommsRouting")
     version: IterationVersion = Field(..., alias="Version")
     name: IterationName = Field(..., alias="Name")
-    iteration_date: str | None = Field(None, alias="IterationDate")
+    iteration_date: IterationDate = Field(..., alias="IterationDate")
     iteration_number: int | None = Field(None, alias="IterationNumber")
-    comms_type: Literal["I", "R"] = Field(..., alias="CommsType")
     approval_minimum: int | None = Field(None, alias="ApprovalMinimum")
     approval_maximum: int | None = Field(None, alias="ApprovalMaximum")
     type: Literal["A", "M", "S"] = Field(..., alias="Type")
-    iteration_cohorts: list[IterationCohort] | None = Field(None, alias="IterationCohorts")
-    iteration_rules: list[IterationRule] | None = Field(None, alias="IterationRules")
+    iteration_cohorts: list[IterationCohort] = Field(..., alias="IterationCohorts")
+    iteration_rules: list[IterationRule] = Field(..., alias="IterationRules")
 
-    model_config = {"populate_by_name": True}
+    model_config = {
+        "populate_by_name": True,
+        "arbitrary_types_allowed": True,
+    }
+
+    @field_validator("iteration_date", mode="before")
+    @classmethod
+    def parse_dates(cls, v: str | date) -> date:
+        if isinstance(v, date):
+            return v
+        return datetime.strptime(v, "%Y%m%d").date()  # noqa: DTZ007
+
+    @field_serializer("iteration_date", when_used="always")
+    @staticmethod
+    def serialize_dates(v: date, _info: SerializationInfo) -> str:
+        return v.strftime("%Y%m%d")
 
 
 class CampaignConfig(BaseModel):
@@ -101,7 +120,7 @@ class CampaignConfig(BaseModel):
     end_date: EndDate = Field(..., alias="EndDate")
     approval_minimum: int | None = Field(None, alias="ApprovalMinimum")
     approval_maximum: int | None = Field(None, alias="ApprovalMaximum")
-    iterations: list[Iteration] | None = Field(None, alias="Iterations")
+    iterations: list[Iteration] = Field(..., alias="Iterations")
 
     model_config = {
         "populate_by_name": True,

src/eligibility_signposting_api/model/rules.py

@@ -1,4 +1,5 @@
 from .eligibility_repo import EligibilityRepo
 from .exceptions import NotFoundError
+from .rules_repo import RulesRepo
 
-__all__ = ["EligibilityRepo", "NotFoundError"]
+__all__ = ["EligibilityRepo", "NotFoundError", "RulesRepo"]

src/eligibility_signposting_api/repos/__init__.py

@@ -1,10 +1,11 @@
 import json
+from collections.abc import Generator
 from typing import Annotated
 
 from botocore.client import BaseClient
 from wireup import Inject, service
 
-from eligibility_signposting_api.model.rules import BucketName, CampaignConfig, CampaignName, Rules
+from eligibility_signposting_api.model.rules import BucketName, CampaignConfig, Rules
 
 
 @service
@@ -18,7 +19,9 @@ def __init__(
         self.s3_client = s3_client
         self.bucket_name = bucket_name
 
-    def get_campaign_config(self, campaign: CampaignName) -> CampaignConfig:
-        response = self.s3_client.get_object(Bucket=self.bucket_name, Key=f"{campaign}.json")
-        body = response["Body"].read()
-        return Rules.model_validate(json.loads(body)).campaign_config
+    def get_campaign_configs(self) -> Generator[CampaignConfig]:
+        campaign_objects = self.s3_client.list_objects(Bucket=self.bucket_name)
+        for campaign_object in campaign_objects["Contents"]:
+            response = self.s3_client.get_object(Bucket=self.bucket_name, Key=f"{campaign_object['Key']}")
+            body = response["Body"].read()
+            yield Rules.model_validate(json.loads(body)).campaign_config

src/eligibility_signposting_api/repos/rules_repo.py

@@ -1,9 +1,13 @@
 import logging
+from datetime import datetime
+from typing import Any
 
+from dateutil.relativedelta import relativedelta
 from wireup import service
 
-from eligibility_signposting_api.model.eligibility import Eligibility, NHSNumber
-from eligibility_signposting_api.repos import EligibilityRepo, NotFoundError
+from eligibility_signposting_api.model.eligibility import EligibilityStatus, NHSNumber
+from eligibility_signposting_api.model.rules import CampaignConfig, IterationRule, RuleAttributeLevel, RuleOperator
+from eligibility_signposting_api.repos import EligibilityRepo, NotFoundError, RulesRepo
 
 logger = logging.getLogger(__name__)
 
@@ -14,23 +18,87 @@ class UnknownPersonError(Exception):
 
 @service
 class EligibilityService:
-    def __init__(self, eligibility_repo: EligibilityRepo) -> None:
+    def __init__(self, eligibility_repo: EligibilityRepo, rules_repo: RulesRepo) -> None:
         super().__init__()
         self.eligibility_repo = eligibility_repo
+        self.rules_repo = rules_repo
 
-    def get_eligibility(self, nhs_number: NHSNumber | None = None) -> Eligibility:
+    def get_eligibility_status(self, nhs_number: NHSNumber | None = None) -> EligibilityStatus:
         if nhs_number:
             try:
-                eligibility_data = self.eligibility_repo.get_eligibility_data(nhs_number)
+                person_data = self.eligibility_repo.get_eligibility_data(nhs_number)
+                campaign_configs = list(self.rules_repo.get_campaign_configs())
                 logger.debug(
-                    "got eligibility_data %r",
-                    eligibility_data,
-                    extra={"eligibility_data": eligibility_data, "nhs_number": nhs_number},
+                    "got person_data for %r",
+                    nhs_number,
+                    extra={
+                        "campaign_configs": [c.model_dump(by_alias=True) for c in campaign_configs],
+                        "person_data": person_data,
+                        "nhs_number": nhs_number,
+                    },
                 )
             except NotFoundError as e:
                 raise UnknownPersonError from e
             else:
                 # TODO: Apply rules here  # noqa: TD002, TD003, FIX002
-                return Eligibility(processed_suggestions=[])
+                return self.evaluate_eligibility(campaign_configs, person_data)
 
         raise UnknownPersonError
+
+    @staticmethod
+    def evaluate_eligibility(
+        campaign_configs: list[CampaignConfig], person_data: list[dict[str, Any]]
+    ) -> EligibilityStatus:
+        eligible, reasons, actions = True, [], []
+        for iteration_rule in [
+            iteration_rule
+            for campaign_config in campaign_configs
+            for iteration in campaign_config.iterations
+            for iteration_rule in iteration.iteration_rules
+        ]:
+            if EligibilityService.evaluate_exclusion(iteration_rule, person_data):
+                eligible = False
+
+        return EligibilityStatus(eligible=eligible, reasons=reasons, actions=actions)
+
+    @staticmethod
+    def evaluate_exclusion(iteration_rule: IterationRule, person_data: list[dict[str, Any]]) -> bool:
+        attribute_value = EligibilityService.get_attribute_value(iteration_rule, person_data)
+        return EligibilityService.evaluate_rule(iteration_rule, attribute_value)
+
+    @staticmethod
+    def get_attribute_value(iteration_rule: IterationRule, person_data: list[dict[str, Any]]) -> Any:  # noqa: ANN401
+        match iteration_rule.attribute_level:
+            case RuleAttributeLevel.PERSON:
+                person: dict[str, Any] | None = next(
+                    (r for r in person_data if r.get("ATTRIBUTE_TYPE", "").startswith("PERSON")), None
+                )
+                attribute_value = person.get(iteration_rule.attribute_name) if person else None
+            case _:
+                msg = f"{iteration_rule.attribute_level} not implemented"
+                raise NotImplementedError(msg)
+        return attribute_value
+
+    @staticmethod
+    def evaluate_rule(iteration_rule: IterationRule, attribute_value: Any) -> bool:  # noqa: PLR0911, ANN401
+        match iteration_rule.operator:
+            case RuleOperator.equals:
+                return attribute_value == iteration_rule.comparator
+            case RuleOperator.ne:
+                return attribute_value != iteration_rule.comparator
+            case RuleOperator.lt:
+                return int(attribute_value) < int(iteration_rule.comparator)
+            case RuleOperator.lte:
+                return int(attribute_value) <= int(iteration_rule.comparator)
+            case RuleOperator.gt:
+                return int(attribute_value) > int(iteration_rule.comparator)
+            case RuleOperator.gte:
+                return int(attribute_value) >= int(iteration_rule.comparator)
+            case RuleOperator.year_gt:
+                attribute_date = datetime.strptime(str(attribute_value), "%Y%m%d") if attribute_value else None  # noqa: DTZ007
+                today = datetime.today()  # noqa: DTZ002
+                cutoff = today + relativedelta(years=int(iteration_rule.comparator))
+                return (attribute_date > cutoff) if attribute_date else False
+            case _:
+                msg = f"{iteration_rule.operator} not implemented"
+                raise NotImplementedError(msg)

src/eligibility_signposting_api/services/eligibility_services.py

@@ -1,14 +1,18 @@
 import logging
 from http import HTTPStatus
 
-from fhir.resources.operationoutcome import OperationOutcome, OperationOutcomeIssue
+from fhir.resources.R4B.bundle import Bundle, BundleEntry
+from fhir.resources.R4B.guidanceresponse import GuidanceResponse
+from fhir.resources.R4B.location import Location
+from fhir.resources.R4B.operationoutcome import OperationOutcome, OperationOutcomeIssue
+from fhir.resources.R4B.requestgroup import RequestGroup
+from fhir.resources.R4B.task import Task
 from flask import Blueprint, make_response, request
 from flask.typing import ResponseReturnValue
 from wireup import Injected
 
-from eligibility_signposting_api.model.eligibility import Eligibility, NHSNumber
+from eligibility_signposting_api.model.eligibility import EligibilityStatus, NHSNumber
 from eligibility_signposting_api.services import EligibilityService, UnknownPersonError
-from eligibility_signposting_api.views.response_models import EligibilityResponse
 
 logger = logging.getLogger(__name__)
 
@@ -20,7 +24,7 @@ def check_eligibility(eligibility_service: Injected[EligibilityService]) -> Resp
     nhs_number = NHSNumber(request.args.get("nhs_number", ""))
     logger.debug("checking nhs_number %r in %r", nhs_number, eligibility_service, extra={"nhs_number": nhs_number})
     try:
-        eligibility = eligibility_service.get_eligibility(nhs_number)
+        eligibility_status = eligibility_service.get_eligibility_status(nhs_number)
     except UnknownPersonError:
         logger.debug("nhs_number %r not found", nhs_number, extra={"nhs_number": nhs_number})
         problem = OperationOutcome(
@@ -32,11 +36,22 @@ def check_eligibility(eligibility_service: Injected[EligibilityService]) -> Resp
                 )  # pyright: ignore[reportCallIssue]
             ]
         )
-        return make_response(problem.model_dump(), HTTPStatus.NOT_FOUND)
+        return make_response(problem.model_dump(by_alias=True), HTTPStatus.NOT_FOUND)
     else:
-        eligibility_response = build_eligibility_response(eligibility)
-        return make_response(eligibility_response.model_dump(), HTTPStatus.OK)
-
-
-def build_eligibility_response(eligibility: Eligibility) -> EligibilityResponse:
-    return EligibilityResponse(processed_suggestions=eligibility.processed_suggestions)
+        bundle = build_bundle(eligibility_status)
+        return make_response(bundle.model_dump(by_alias=True), HTTPStatus.OK)
+
+
+def build_bundle(_eligibility_status: EligibilityStatus) -> Bundle:
+    return Bundle(  # pyright: ignore[reportCallIssue]
+        id="dummy-bundle",
+        type="collection",
+        entry=[
+            BundleEntry(  # pyright: ignore[reportCallIssue]
+                resource=GuidanceResponse(id="dummy-guidance-response", status="requested", moduleCodeableConcept={})  # pyright: ignore[reportCallIssue]
+            ),
+            BundleEntry(resource=RequestGroup(id="dummy-request-group", intent="proposal", status="requested")),  # pyright: ignore[reportCallIssue]
+            BundleEntry(resource=Task(id="dummy-task", intent="proposal", status="requested")),  # pyright: ignore[reportCallIssue]
+            BundleEntry(resource=Location(id="dummy-location")),  # pyright: ignore[reportCallIssue]
+        ],
+    )