Merge pull request #610 from NHSDigital/spike/ELI-619-caching

[ELI-619] - cache campaign configs unless we send a reserved consumer id

Author: TOEL2

poetry.lock

@@ -35,6 +35,7 @@ pyhamcrest = "^2.1.0"
 boto3 = "^1.40.57"
 botocore = "^1.40.76"
 aws-xray-sdk = "2.15.0"
+cachetools = "^7.0.1"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.14.10"

pyproject.toml

@@ -1,3 +1,4 @@
+import os
 from typing import Literal
 
 URL_PREFIX = "patient-check"
@@ -6,3 +7,5 @@
 CONSUMER_ID = "NHSE-Product-ID"
 ALLOWED_CONDITIONS = Literal["COVID", "FLU", "MMR", "RSV"]
 CONSUMER_MAPPING_FILE_NAME = "consumer_mapping_config.json"
+
+CACHE_TTL_SECONDS = int(os.getenv("CONFIG_CACHE_TTL_SECONDS", "1800"))

src/eligibility_signposting_api/config/constants.py

@@ -1,15 +1,22 @@
 import json
+import logging
 from collections.abc import Generator
 from typing import Annotated, NewType
 
 from aws_xray_sdk.core import xray_recorder
 from botocore.client import BaseClient
+from cachetools import TTLCache
 from wireup import Inject, service
 
+from eligibility_signposting_api.config.constants import CACHE_TTL_SECONDS
 from eligibility_signposting_api.model.campaign_config import CampaignConfig, Rules
 
 BucketName = NewType("BucketName", str)
 
+logger = logging.getLogger(__name__)
+
+campaign_config_cache: TTLCache[str, list[CampaignConfig]] = TTLCache(maxsize=1, ttl=CACHE_TTL_SECONDS)
+
 
 @service
 class CampaignRepo:
@@ -26,12 +33,43 @@ def __init__(
         self.s3_client = s3_client
         self.bucket_name = bucket_name
 
-    def get_campaign_configs(self) -> Generator[CampaignConfig]:
+    def get_campaign_configs(self, consumer_id: str) -> Generator[CampaignConfig]:
+        bypass = "test-" in consumer_id
+        cache_key = "all_campaigns"
+        cached = None if bypass else campaign_config_cache.get(cache_key)
+
         with xray_recorder.in_subsegment("CampaignRepo.get_campaign_configs"):
+            if cached is not None:
+                logger.info("Using cached campaign configs")
+                yield from cached
+                return
+
+            logger.info(
+                "Refreshing campaign configs from S3 (consumer_id=%s, ttl_seconds=%s)",
+                consumer_id,
+                CACHE_TTL_SECONDS,
+            )
+            configs = self._load_campaign_configs_from_s3()
+
+            if not bypass:
+                campaign_config_cache[cache_key] = configs
+
+            yield from configs
+
+    def _load_campaign_configs_from_s3(self) -> list[CampaignConfig]:
+        campaign_configs: list[CampaignConfig] = []
+
+        with xray_recorder.in_subsegment("CampaignRepo.load_campaign_configs_from_s3"):
             with xray_recorder.in_subsegment("list_objects"):
                 campaign_objects = self.s3_client.list_objects(Bucket=self.bucket_name)
+
             with xray_recorder.in_subsegment("get_objects"):
-                for campaign_object in campaign_objects["Contents"]:
-                    response = self.s3_client.get_object(Bucket=self.bucket_name, Key=f"{campaign_object['Key']}")
+                for campaign_object in campaign_objects.get("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
+                    campaign_configs.append(Rules.model_validate(json.loads(body)).campaign_config)
+
+        return campaign_configs

src/eligibility_signposting_api/repos/campaign_repo.py

@@ -50,7 +50,7 @@ def get_eligibility_status(
             except NotFoundError as e:
                 raise UnknownPersonError from e
             else:
-                campaign_configs: list[CampaignConfig] = list(self.campaign_repo.get_campaign_configs())
+                campaign_configs: list[CampaignConfig] = list(self.campaign_repo.get_campaign_configs(consumer_id))
                 permitted_campaign_configs = self.__collect_permitted_campaign_configs(
                     campaign_configs, ConsumerId(consumer_id)
                 )

src/eligibility_signposting_api/services/eligibility_services.py

@@ -27,7 +27,6 @@
     Status.not_actionable: eligibility_response.Status.not_actionable,
     Status.not_eligible: eligibility_response.Status.not_eligible,
 }
-
 logger = logging.getLogger(__name__)
 
 eligibility_blueprint = Blueprint("eligibility", __name__)

src/eligibility_signposting_api/views/eligibility.py

@@ -33,6 +33,7 @@ services:
       - FIREHOSE_ENDPOINT=http://moto-server:5000
       - KINESIS_ENDPOINT=http://moto-server:5000
       - LOG_LEVEL=INFO
+      #- ENVIRONMENT=dev
     entrypoint: /bin/sh
     command:
       - "-c"

tests/docker-compose.mock_aws.yml

@@ -33,7 +33,7 @@
 from eligibility_signposting_api.model.consumer_mapping import ConsumerCampaign, ConsumerId, ConsumerMapping
 from eligibility_signposting_api.processors.hashing_service import HashingService, HashSecretName
 from eligibility_signposting_api.repos import SecretRepo
-from eligibility_signposting_api.repos.campaign_repo import BucketName
+from eligibility_signposting_api.repos.campaign_repo import BucketName, campaign_config_cache
 from eligibility_signposting_api.repos.person_repo import TableName
 from tests.fixtures.builders.model import rule
 from tests.fixtures.builders.model.rule import RulesMapperFactory
@@ -103,6 +103,11 @@ def moto_server(request: pytest.FixtureRequest) -> URL:
     return url
 
 
+@pytest.fixture(autouse=True)
+def clear_cache():
+    campaign_config_cache.clear()
+
+
 def is_responsive(url: URL) -> bool:
     try:
         response = httpx.get(str(url))
@@ -1304,7 +1309,7 @@ def campaign_configs(request, s3_client: BaseClient, rules_bucket: BucketName) -
 
 @pytest.fixture(scope="class")
 def consumer_id() -> ConsumerId:
-    return ConsumerId("23-mic7heal-jor6don")
+    return ConsumerId("test-23-mic7heal-jor6don")
 
 
 def create_and_put_consumer_mapping_in_s3(

tests/integration/conftest.py

@@ -27,7 +27,7 @@ def test_get_campaign_config(s3_client: BaseClient, rules_bucket: BucketName, ca
     repo = CampaignRepo(s3_client, rules_bucket)
 
     # When
-    actual = list(repo.get_campaign_configs())
+    actual = list(repo.get_campaign_configs("consumer_id"))
 
     # Then
     assert_that(

tests/integration/repo/test_campaign_repo.py

@@ -0,0 +1,94 @@
+import io
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+from eligibility_signposting_api.repos.campaign_repo import BucketName, CampaignRepo, campaign_config_cache
+from tests.fixtures.builders.model.rule import CampaignConfigFactory
+
+
+def make_s3_body(payload: dict):
+    return {"Body": io.BytesIO(json.dumps(payload).encode("utf-8"))}
+
+
+class TestCampaignRepo:
+    @pytest.fixture(autouse=True)
+    def clear_cache(self):
+        campaign_config_cache.clear()
+
+    @pytest.fixture
+    def mock_s3_client(self):
+        return MagicMock()
+
+    @pytest.fixture
+    def repo(self, mock_s3_client):
+        return CampaignRepo(
+            s3_client=mock_s3_client,
+            bucket_name=BucketName("test-bucket"),
+        )
+
+    @pytest.fixture
+    def rules_payload(self):
+        campaign_config = CampaignConfigFactory.build()
+        return {"campaign_config": campaign_config.model_dump(mode="json")}
+
+    def test_get_campaign_configs_loads_from_s3(self, repo, mock_s3_client, rules_payload):
+        mock_s3_client.list_objects.return_value = {"Contents": [{"Key": "rsv.json"}]}
+        mock_s3_client.get_object.return_value = make_s3_body(rules_payload)
+
+        result = list(repo.get_campaign_configs("consumer_id"))
+
+        assert len(result) == 1
+        assert result[0].id == rules_payload["campaign_config"]["id"]
+
+        mock_s3_client.list_objects.assert_called_once_with(Bucket="test-bucket")
+        mock_s3_client.get_object.assert_called_once_with(
+            Bucket="test-bucket",
+            Key="rsv.json",
+        )
+
+    def test_get_campaign_configs_uses_cache_within_ttl(
+        self,
+        repo,
+        mock_s3_client,
+    ):
+        first_config = CampaignConfigFactory.build(version=1)
+
+        mock_s3_client.list_objects.return_value = {"Contents": [{"Key": "rsv.json"}]}
+        mock_s3_client.get_object.return_value = make_s3_body({"campaign_config": first_config.model_dump(mode="json")})
+
+        first = list(repo.get_campaign_configs("consumer_id"))
+        second = list(repo.get_campaign_configs("consumer_id"))
+
+        assert first[0].version == 1
+        assert second[0].version == 1
+        assert mock_s3_client.list_objects.call_count == 1
+        assert mock_s3_client.get_object.call_count == 1
+
+    def test_get_campaign_configs_refreshes_after_cache_is_cleared(
+        self,
+        repo,
+        mock_s3_client,
+    ):
+        first_config = CampaignConfigFactory.build(version=1)
+        second_config = CampaignConfigFactory.build(version=2)
+
+        mock_s3_client.list_objects.return_value = {"Contents": [{"Key": "rsv.json"}]}
+        mock_s3_client.get_object.side_effect = [
+            make_s3_body({"campaign_config": first_config.model_dump(mode="json")}),
+            make_s3_body({"campaign_config": second_config.model_dump(mode="json")}),
+        ]
+
+        first = list(repo.get_campaign_configs("consumer_id"))
+        second = list(repo.get_campaign_configs("consumer_id"))
+        campaign_config_cache.clear()
+        third = list(repo.get_campaign_configs("test-consumer-1"))
+
+        expected_call_count = 2
+
+        assert first[0].version == first_config.version
+        assert second[0].version == first_config.version
+        assert third[0].version == second_config.version
+        assert mock_s3_client.list_objects.call_count == expected_call_count
+        assert mock_s3_client.get_object.call_count == expected_call_count