ELI-578 consumer_id - campaign_config mapping

Author: Karthikeyannhs

src/eligibility_signposting_api/model/consumer_mapping.py

@@ -0,0 +1,11 @@
+from typing import NewType
+
+from pydantic import RootModel
+
+from eligibility_signposting_api.model.campaign_config import CampaignID
+
+ConsumerId = NewType("ConsumerId", str)
+
+class ConsumerMapping(RootModel[dict[str, list[CampaignID]]]):
+    def get(self, key: str, default: list[CampaignID] | None = None) -> list[CampaignID] | None:
+        return self.root.get(key, default)

src/eligibility_signposting_api/repos/consumer_mapping_repo.py

@@ -0,0 +1,32 @@
+import json
+from typing import Annotated, NewType
+
+from botocore.client import BaseClient
+from wireup import Inject, service
+
+from eligibility_signposting_api.model.campaign_config import CampaignID
+from eligibility_signposting_api.model.consumer_mapping import ConsumerMapping, ConsumerId
+
+BucketName = NewType("BucketName", str)
+
+
+@service
+class ConsumerMappingRepo:
+    """Repository class for Campaign Rules, which we can use to calculate a person's eligibility for vaccination.
+
+    These rules are stored as JSON files in AWS S3."""
+
+    def __init__(
+        self,
+        s3_client: Annotated[BaseClient, Inject(qualifier="s3")],
+        bucket_name: Annotated[BucketName, Inject(param="rules_bucket_name")],
+    ) -> None:
+        super().__init__()
+        self.s3_client = s3_client
+        self.bucket_name = bucket_name
+
+    def get_sanctioned_campaign_ids(self, consumer_id: ConsumerId) -> list[CampaignID] | None:
+        consumer_mappings = self.s3_client.list_objects(Bucket=self.bucket_name)["Contents"][0]
+        response = self.s3_client.get_object(Bucket=self.bucket_name, Key=f"{consumer_mappings['Key']}")
+        body = response["Body"].read()
+        return ConsumerMapping.model_validate(json.loads(body)).get(consumer_id)

src/eligibility_signposting_api/services/eligibility_services.py

@@ -3,7 +3,9 @@
 from wireup import service
 
 from eligibility_signposting_api.model import eligibility_status
+from eligibility_signposting_api.model.campaign_config import CampaignConfig
 from eligibility_signposting_api.repos import CampaignRepo, NotFoundError, PersonRepo
+from eligibility_signposting_api.repos.consumer_mapping_repo import ConsumerMappingRepo
 from eligibility_signposting_api.services.calculators import eligibility_calculator as calculator
 
 logger = logging.getLogger(__name__)
@@ -24,28 +26,38 @@ def __init__(
         person_repo: PersonRepo,
         campaign_repo: CampaignRepo,
         calculator_factory: calculator.EligibilityCalculatorFactory,
+        consumer_mapping_repo: ConsumerMappingRepo
     ) -> None:
         super().__init__()
         self.person_repo = person_repo
         self.campaign_repo = campaign_repo
         self.calculator_factory = calculator_factory
+        self.consumer_mapping = consumer_mapping_repo
 
     def get_eligibility_status(
         self,
         nhs_number: eligibility_status.NHSNumber,
         include_actions: str,
         conditions: list[str],
         category: str,
+        consumer_id: str,
     ) -> eligibility_status.EligibilityStatus:
         """Calculate a person's eligibility for vaccination given an NHS number."""
         if nhs_number:
             try:
                 person_data = self.person_repo.get_eligibility_data(nhs_number)
                 campaign_configs = list(self.campaign_repo.get_campaign_configs())
+                consumer_mappings = self.consumer_mapping.get_sanctioned_campaign_ids(consumer_id)
+                sanctioned_campaign_ids: list[CampaignConfig] = [
+                    campaign for campaign in campaign_configs
+                    if campaign.id in consumer_mappings
+                ]
+
             except NotFoundError as e:
                 raise UnknownPersonError from e
             else:
-                calc: calculator.EligibilityCalculator = self.calculator_factory.get(person_data, campaign_configs)
+                calc: calculator.EligibilityCalculator = self.calculator_factory.get(person_data,
+                                                                                     sanctioned_campaign_ids)
                 return calc.get_eligibility_status(include_actions, conditions, category)
 
         raise UnknownPersonError  # pragma: no cover

src/eligibility_signposting_api/views/eligibility.py

@@ -54,6 +54,7 @@ def check_eligibility(
             query_params["includeActions"],
             query_params["conditions"],
             query_params["category"],
+            request.headers.get("X-Correlation-ID"),
         )
     except UnknownPersonError:
         return handle_unknown_person_error(nhs_number)