New: [AEA-6067] - Reimplement expireAt calculation

.github/dependabot.yaml

@@ -9,7 +9,6 @@ updates:
       time: "18:00" # UTC
     commit-message:
       prefix: "Upgrade: [dependabot] - "
-      include: "scope"
 
   - package-ecosystem: "pip"
     directory: "/"
@@ -21,4 +20,3 @@ updates:
     open-pull-requests-limit: 20
     commit-message:
       prefix: "Upgrade: [dependabot] - "
-      include: "scope"

.github/workflows/ci.yaml

@@ -26,4 +26,5 @@ jobs:
     with:
       asdfVersion: ${{ needs.get_asdf_version.outputs.version }}
       reinstall_poetry: true
-      run_sonar: false
+    secrets:
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

.github/workflows/pull_request.yaml

@@ -30,7 +30,8 @@ jobs:
     with:
       asdfVersion: ${{ needs.get_asdf_version.outputs.version }}
       reinstall_poetry: true
-      run_sonar: false
+    secrets:
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
 
   tag_release:
     name: Tag Release (Dry Run)

.github/workflows/release.yaml

@@ -24,7 +24,8 @@ jobs:
     with:
       asdfVersion: ${{ needs.get_asdf_version.outputs.version }}
       reinstall_poetry: true
-      run_sonar: false
+    secrets:
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
 
   get_next_version:
     name: Get Next Version Number for Poetry

.trivyignore

@@ -0,0 +1,5 @@
+# urllib3 - can't upgrade due to spine version restraints
+CVE-2025-66418
+CVE-2025-66471
+CVE-2026-21441
+CVE-2025-50181

src/eps_spine_shared/common/dynamodb_datastore.py

@@ -27,8 +27,8 @@
 from eps_spine_shared.logger import EpsLogger
 from eps_spine_shared.nhsfundamentals.timeutilities import (
     TimeFormats,
-    convertSpineDate,
-    timeNowAsString,
+    convert_spine_date,
+    time_now_as_string,
 )
 
 
@@ -78,6 +78,7 @@ class EpsDynamoDbDataStore:
     NOTIFICATION_PREFIX = "Notification_"
     STORE_TIME_DOC_REF_TITLE_PREFIX = "NominatedReleaseRequestMsgRef"
     DEFAULT_EXPIRY_DAYS = 56
+    MAX_NEXT_ACTIVITY_DATE = "99991231"
 
     def __init__(
         self,
@@ -134,6 +135,41 @@ def get_expire_at(self, delta, from_datetime=None):
 
         return int((from_datetime + delta).timestamp())
 
+    def calculate_record_expire_at(
+        self, next_activity, next_activity_date_str, creation_datetime_string
+    ):
+        """
+        If the record next activity is delete or purge, use the nextActivity and nextActivityDate
+        to calculate its expireAt (ttl) value, otherwise fall-back to the default of 18 months.
+        """
+        creation_datetime = convert_spine_date(
+            creation_datetime_string, TimeFormats.STANDARD_DATE_TIME_FORMAT
+        )
+        creation_datetime_utc = datetime.combine(
+            creation_datetime.date(), creation_datetime.time(), timezone.utc
+        )
+        default_expire_at = self.get_expire_at(relativedelta(months=18), creation_datetime_utc)
+
+        if (
+            next_activity.lower() not in ["delete", "purge"]
+            or not next_activity_date_str
+            or next_activity_date_str == self.MAX_NEXT_ACTIVITY_DATE
+        ):
+            return default_expire_at
+
+        delta = relativedelta() if next_activity.lower() == "purge" else relativedelta(months=12)
+
+        next_activity_datetime = convert_spine_date(
+            next_activity_date_str, TimeFormats.STANDARD_DATE_FORMAT
+        )
+        next_activity_datetime_utc = datetime.combine(
+            next_activity_datetime.date(), next_activity_datetime.time(), timezone.utc
+        )
+
+        next_activity_expire_at = self.get_expire_at(delta, next_activity_datetime_utc)
+
+        return min(next_activity_expire_at, default_expire_at)
+
     def build_document(self, internal_id, document, index):
         """
         Build EPS Document object to be inserted into DynamoDB.
@@ -182,6 +218,22 @@ def convert_index_keys_to_lower_case(self, index):
             return index
         return {key.lower(): index[key] for key in index}
 
+    def parse_next_activity_nad(self, indexes):
+        """
+        Split nextActivityNAD string into sharded next activity and its date.
+        """
+        next_activity_nad = indexes["nextActivityNAD_bin"][0]
+        next_activity_nad_split = next_activity_nad.split("_")
+
+        next_activity = next_activity_nad_split[0]
+        next_activity_date_str = (
+            next_activity_nad_split[1] if len(next_activity_nad_split) == 2 else None
+        )
+
+        shard = randint(1, NEXT_ACTIVITY_DATE_PARTITIONS)
+
+        return next_activity, shard, next_activity_date_str
+
     def build_record(self, prescription_id, record, record_type, indexes):
         """
         Build EPS Record object to be inserted into DynamoDB.
@@ -192,36 +244,34 @@ def build_record(self, prescription_id, record, record_type, indexes):
             indexes = record["indexes"]
         instances = record["instances"].values()
 
-        next_activity_nad = indexes["nextActivityNAD_bin"][0]
-        next_activity_nad_split = next_activity_nad.split("_")
-        next_activity = next_activity_nad_split[0]
-        next_activity_is_purge = next_activity.lower() == "purge"
-
-        next_activity_shard = randint(1, NEXT_ACTIVITY_DATE_PARTITIONS)
-        sharded_next_activity = f"{next_activity}.{next_activity_shard}"
+        next_activity, shard, next_activity_date_str = self.parse_next_activity_nad(indexes)
 
         scn = record["SCN"]
 
         compressed_record = zlib.compress(simplejson.dumps(record).encode("utf-8"))
 
+        creation_datetime_string = record["prescription"]["prescriptionTime"]
+
+        expire_at = self.calculate_record_expire_at(
+            next_activity, next_activity_date_str, creation_datetime_string
+        )
+
         item = {
             Key.PK.name: record_key,
             Key.SK.name: SortKey.RECORD.value,
             ProjectedAttribute.BODY.name: compressed_record,
-            Attribute.NEXT_ACTIVITY.name: sharded_next_activity,
+            Attribute.NEXT_ACTIVITY.name: f"{next_activity}.{shard}",
             ProjectedAttribute.SCN.name: scn,
             ProjectedAttribute.INDEXES.name: self.convert_index_keys_to_lower_case(indexes),
+            ProjectedAttribute.EXPIRE_AT.name: expire_at,
         }
-        if len(next_activity_nad_split) == 2:
-            item[Attribute.NEXT_ACTIVITY_DATE.name] = next_activity_nad_split[1]
+        if next_activity_date_str:
+            item[Attribute.NEXT_ACTIVITY_DATE.name] = next_activity_date_str
 
-        if next_activity_is_purge:
+        if next_activity.lower() == "purge":
             return item
 
-        # POC - Leverage methods in PrescriptionRecord to get some/all of these.
-        creation_datetime_string = record["prescription"]["prescriptionTime"]
         nhs_number = record["patient"]["nhsNumber"]
-
         prescriber_org = record["prescription"]["prescribingOrganization"]
 
         statuses = list(set([instance["prescriptionStatus"] for instance in instances]))
@@ -240,21 +290,12 @@ def build_record(self, prescription_id, record, record_type, indexes):
 
         nominated_pharmacy = record.get("nomination", {}).get("nominatedPerformer")
 
-        creation_datetime = convertSpineDate(
-            creation_datetime_string, TimeFormats.STANDARD_DATE_TIME_FORMAT
-        )
-        creation_datetime_utc = datetime.combine(
-            creation_datetime.date(), creation_datetime.time(), timezone.utc
-        )
-        expire_at = self.get_expire_at(relativedelta(months=18), creation_datetime_utc)
-
         item_update = {
             Attribute.CREATION_DATETIME.name: creation_datetime_string,
             Attribute.NHS_NUMBER.name: nhs_number,
             Attribute.PRESCRIBER_ORG.name: prescriber_org,
             ProjectedAttribute.STATUS.name: status,
             Attribute.IS_READY.name: int(is_ready),
-            ProjectedAttribute.EXPIRE_AT.name: expire_at,
         }
         if dispenser_org:
             item[Attribute.DISPENSER_ORG.name] = dispenser_org
@@ -285,7 +326,7 @@ def insert_eps_work_list(self, internal_id, message_id, work_list, index=None):
         """
         Insert EPS WorkList object into the configured table.
         """
-        work_list_indexes = {self.INDEX_WORKLISTDATE: [timeNowAsString()]}
+        work_list_indexes = {self.INDEX_WORKLISTDATE: [time_now_as_string()]}
         if index:
             work_list_indexes = index
 
@@ -599,7 +640,7 @@ def store_batch_claim(self, internal_id, batch_claim_original):
         claim_id_index_terms = batch_claim["Claim ID List"]
         handle_time_index_term = batch_claim["Handle Time"]
         sequence_number = batch_claim["Sequence Number"]
-        index_scn_value = f"{timeNowAsString()}|{sequence_number}"
+        index_scn_value = f"{time_now_as_string()}|{sequence_number}"
 
         nwssp = "Nwssp Sequence Number" in batch_claim
         nwssp_sequence_number = batch_claim.get("Nwssp Sequence Number")

src/eps_spine_shared/common/dynamodb_index.py

@@ -154,7 +154,6 @@ def build_terms(self, items, index_name, term_regex):
         """
         Build terms from items returned by the index query.
         """
-        # POC - Project the body into the index and do away with 'terms' altogether.
         terms = []
         for item in items:
             index_terms = item.get(ProjectedAttribute.INDEXES.name, {}).get(index_name.lower())
@@ -163,7 +162,6 @@ def build_terms(self, items, index_name, term_regex):
             [
                 terms.append((index_term, item[Key.PK.name]))
                 for index_term in index_terms
-                # POC - term_regex can be replaced by filter expressions for status and releaseVersion.
                 if ((not term_regex) or re.search(term_regex, index_term))
             ]
         return terms

src/eps_spine_shared/common/indexes.py

@@ -1,6 +1,6 @@
 from eps_spine_shared.errors import EpsSystemError
 from eps_spine_shared.logger import EpsLogger
-from eps_spine_shared.nhsfundamentals.timeutilities import timeNowAsString
+from eps_spine_shared.nhsfundamentals.timeutilities import time_now_as_string
 
 INDEX_NHSNUMBER_DATE = "nhsNumberDate_bin"
 INDEX_NHSNUMBER_PRDATE = "nhsNumberPrescriberDate_bin"
@@ -243,4 +243,4 @@ def _add_delta_index(self, eps_record, index_dict):
         """
         See build_indexes
         """
-        index_dict[INDEX_DELTA] = [timeNowAsString() + SEPERATOR + str(eps_record.get_scn())]
+        index_dict[INDEX_DELTA] = [time_now_as_string() + SEPERATOR + str(eps_record.get_scn())]

src/eps_spine_shared/common/prescription/line_item.py

@@ -85,17 +85,17 @@ def expire(self, parent_prescription):
 
         :type parent_prescription: PrescriptionRecord
         """
-        currentStatus = self.status
-        if currentStatus not in LineItemStatus.EXPIRY_IMMUTABLE_STATES:
-            newStatus = LineItemStatus.EXPIRY_LOOKUP[currentStatus]
-            self.update_status(newStatus)
+        current_status = self.status
+        if current_status not in LineItemStatus.EXPIRY_IMMUTABLE_STATES:
+            new_status = LineItemStatus.EXPIRY_LOOKUP[current_status]
+            self.update_status(new_status)
             parent_prescription.logObject.write_log(
                 "EPS0072b",
                 None,
                 {
-                    "internalID": parent_prescription.internalID,
+                    "internalID": parent_prescription.internal_id,
                     "lineItemChanged": self.id,
-                    "previousStatus": currentStatus,
-                    "newStatus": newStatus,
+                    "previousStatus": current_status,
+                    "newStatus": new_status,
                 },
             )

src/eps_spine_shared/common/prescription/record.py

@@ -16,7 +16,7 @@
 )
 from eps_spine_shared.logger import EpsLogger
 from eps_spine_shared.nhsfundamentals.timeutilities import TimeFormats
-from eps_spine_shared.spinecore.baseutilities import handleEncodingOddities, quoted
+from eps_spine_shared.spinecore.baseutilities import handle_encoding_oddities, quoted
 from eps_spine_shared.spinecore.changelog import PrescriptionsChangeLogProcessor
 
 
@@ -158,7 +158,7 @@ def add_event_to_change_log(self, message_id, event_log):
         event_log[PrescriptionsChangeLogProcessor.SCN] = self.get_scn()
         length_before = len(self.prescription_record.get(fields.FIELD_CHANGE_LOG, []))
         try:
-            PrescriptionsChangeLogProcessor.updateChangeLog(
+            PrescriptionsChangeLogProcessor.update_change_log(
                 self.prescription_record, event_log, message_id, self.SCN_MAX
             )
         except Exception as e:  # noqa: BLE001
@@ -901,15 +901,15 @@ def _create_cancellation_summary_dict(
             _cancellation_reasons = str(cancellation_status)
 
             _cancellation_id = _cancellation.get(fields.FIELD_CANCELLATION_ID, [])
-            _scn = PrescriptionsChangeLogProcessor.getSCN(
+            _scn = PrescriptionsChangeLogProcessor.get_scn(
                 self.prescription_record["changeLog"].get(_cancellation_id, {})
             )
             for _cancellation_reason in _cancellation.get(fields.FIELD_REASONS, []):
                 _cancellation_text = _cancellation_reason.split(":")[1].strip()
                 if _subsequent_reason:
                     _cancellation_reasons += "; "
                 _subsequent_reason = True
-                _cancellation_reasons += str(handleEncodingOddities(_cancellation_text))
+                _cancellation_reasons += str(handle_encoding_oddities(_cancellation_text))
 
             if (
                 _cancellation.get(fields.FIELD_CANCELLATION_TARGET) == "Prescription"

src/eps_spine_shared/nhsfundamentals/timeutilities.py

@@ -34,54 +34,54 @@ class TimeFormats:
 }
 
 
-def _guessCommonDateTimeFormat(timeString, raiseErrorIfUnknown=False):
+def _guess_common_datetime_format(time_string, raise_error_if_unknown=False):
     """
     Guess the date time format from the commonly used list
 
     Args:
-        timeString (str):
+        time_string (str):
             The datetime string to try determine the format of.
-        raiseErrorIfUnknown (bool):
+        raise_error_if_unknown (bool):
             Determines the action when the format cannot be determined.
             False (default) will return None, True will raise an error.
     """
-    _format = None
-    if len(timeString) == 19:
+    time_format = None
+    if len(time_string) == 19:
         try:
-            datetime.strptime(timeString, TimeFormats.EBXML_FORMAT)
-            _format = TimeFormats.EBXML_FORMAT
+            datetime.strptime(time_string, TimeFormats.EBXML_FORMAT)
+            time_format = TimeFormats.EBXML_FORMAT
         except ValueError:
-            _format = TimeFormats.STANDARD_DATE_TIME_UTC_ZONE_FORMAT
+            time_format = TimeFormats.STANDARD_DATE_TIME_UTC_ZONE_FORMAT
     else:
-        _format = _TIMEFORMAT_LENGTH_MAP.get(len(timeString), None)
+        time_format = _TIMEFORMAT_LENGTH_MAP.get(len(time_string), None)
 
-    if not _format and raiseErrorIfUnknown:
-        raise ValueError("Could not determine datetime format of '{}'".format(timeString))
+    if not time_format and raise_error_if_unknown:
+        raise ValueError("Could not determine datetime format of '{}'".format(time_string))
 
-    return _format
+    return time_format
 
 
-def convertSpineDate(dateString, dateFormat=None):
+def convert_spine_date(date_string, date_format=None):
     """
     Try to convert a Spine date using the passed format - if it fails - try the most
     appropriate
     """
-    if dateFormat:
+    if date_format:
         try:
-            dateObject = datetime.strptime(dateString, dateFormat)
-            return dateObject
+            date_object = datetime.strptime(date_string, date_format)
+            return date_object
         except ValueError:
             pass
 
-    dateFormat = _guessCommonDateTimeFormat(dateString, raiseErrorIfUnknown=True)
-    return datetime.strptime(dateString, dateFormat)
+    date_format = _guess_common_datetime_format(date_string, raise_error_if_unknown=True)
+    return datetime.strptime(date_string, date_format)
 
 
-def timeNowAsString(_dateFormat=TimeFormats.STANDARD_DATE_TIME_FORMAT):
+def time_now_as_string(date_format=TimeFormats.STANDARD_DATE_TIME_FORMAT):
     """
     Return the current date and time as a string in standard format
     """
-    return now().strftime(_dateFormat)
+    return now().strftime(date_format)
 
 
 def now():