[PRMP-781] Fixed Values Metadata Processor (#908)

Co-authored-by: Robert Gaskin <[email protected]>

Author: SWhyteAnswer

lambdas/handlers/bulk_upload_metadata_processor_handler.py

@@ -3,11 +3,13 @@
     BulkUploadMetadataProcessorService,
     get_formatter_service,
 )
+from services.metadata_mapping_validator_service import MetadataMappingValidatorService
 from utils.audit_logging_setup import LoggingService
 from utils.decorators.ensure_env_var import ensure_environment_variables
 from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
 from utils.decorators.override_error_check import override_error_check
 from utils.decorators.set_audit_arg import set_request_context_for_logging
+from utils.exceptions import BulkUploadMetadataException
 
 logger = LoggingService(__name__)
 
@@ -48,4 +50,16 @@ def lambda_handler(event, _context):
         f"Starting metadata processing for practice directory: {practice_directory}"
     )
 
+    fixed_values = event.get("fixedValues", {})
+
+    validator_service = MetadataMappingValidatorService()
+    validator_service.validate_fixed_values(
+        fixed_values, remappings)
+
+    metadata_formatter_service = formatter_service_class(practice_directory)
+    metadata_service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=metadata_formatter_service,
+        metadata_heading_remap=remappings,
+        fixed_values=fixed_values
+    )
     metadata_service.process_metadata()

lambdas/services/bulk_upload_metadata_processor_service.py

@@ -56,6 +56,7 @@ def __init__(
         self,
         metadata_formatter_service: MetadataPreprocessorService,
         metadata_heading_remap: dict,
+        fixed_values: dict = None,
     ):
         self.staging_bucket_name = os.getenv("STAGING_STORE_BUCKET_NAME")
         self.metadata_queue_url = os.getenv("METADATA_SQS_QUEUE_URL")
@@ -67,6 +68,7 @@ def __init__(
         self.virus_scan_service = get_virus_scan_service()
 
         self.metadata_heading_remap = metadata_heading_remap
+        self.fixed_values = fixed_values or {}
 
         self.temp_download_dir = tempfile.mkdtemp()
         self.practice_directory = metadata_formatter_service.practice_directory
@@ -148,10 +150,10 @@ def csv_to_sqs_metadata(self, csv_file_path: str) -> list[StagingSqsMetadata]:
 
         validated_rows, rejected_rows, rejected_reasons = (
             self.metadata_mapping_validator_service.validate_and_normalize_metadata(
-                records, self.metadata_heading_remap
+                records, self.fixed_values, self.metadata_heading_remap
             )
         )
-        if rejected_reasons:
+        if rejected_reasons: 
             for reason in rejected_reasons:
                 logger.warning(f"Rejected due to: {reason['REASON']}")
 
@@ -177,6 +179,10 @@ def process_metadata_row(
     ) -> None:
         """Validate individual file metadata and attach to patient group."""
         file_metadata = MetadataFile.model_validate(row)
+
+        if self.fixed_values:
+            file_metadata = self.apply_fixed_values(file_metadata)
+
         nhs_number, ods_code = self.extract_patient_info(file_metadata)
 
         try:
@@ -191,6 +197,17 @@ def process_metadata_row(
                                   sqs_metadata.file_path.lstrip("/"))
         patients[(nhs_number, ods_code)].append(sqs_metadata)
 
+    def apply_fixed_values(self, file_metadata: MetadataFile) -> MetadataFile:
+        
+        metadata_dict = file_metadata.model_dump(by_alias=True)
+
+        for field_name, fixed_value in self.fixed_values.items():
+            metadata_dict[field_name] = fixed_value
+            logger.info(
+                f"Applied fixed value for field '{field_name}': '{fixed_value}'")
+            
+        return MetadataFile.model_validate(metadata_dict)
+
     @staticmethod
     def convert_to_sqs_metadata(
             file: MetadataFile, stored_file_name: str

lambdas/services/bulk_upload_service.py

@@ -74,7 +74,9 @@ def process_message_queue(self, records: list):
                 logger.info(
                     "Cannot validate patient due to PDS responded with Too Many Requests"
                 )
-                logger.info("Cannot process for now due to PDS rate limit reached.")
+                logger.info(
+                    "Cannot process for now due to PDS rate limit reached."
+                )
                 logger.info(
                     "All remaining messages in this batch will be returned to sqs queue to retry later."
                 )

lambdas/services/metadata_mapping_validator_service.py

@@ -3,12 +3,15 @@
 from models.staging_metadata import MetadataFile
 from pydantic import BaseModel, ConfigDict, Field, create_model
 from utils.audit_logging_setup import LoggingService
+from utils.exceptions import BulkUploadMetadataException
 
 logger = LoggingService(__name__)
 
 
 class MetadataMappingValidatorService:
     model_aliases = {f.alias for f in MetadataFile.model_fields.values() if f.alias}
+    # Fields that should never be set as fixed by the user
+    PROTECTED_FIELDS = {"FILEPATH", "NHS-NO"}
 
     def create_metadata_model(self, alias_map) -> Type[BaseModel]:
         if not alias_map:
@@ -42,7 +45,7 @@ def create_metadata_model(self, alias_map) -> Type[BaseModel]:
         )
         return dynamic_model
 
-    def validate_and_normalize_metadata(self, records: list[dict], remappings: dict):
+    def validate_and_normalize_metadata(self, records: list[dict],fixed_values: dict, remappings: dict):
         model = self.create_metadata_model(remappings)
         validated_rows, rejected_rows, rejected_reasons = [], [], []
         required_fields = [
@@ -53,8 +56,12 @@ def validate_and_normalize_metadata(self, records: list[dict], remappings: dict)
 
         for row in records:
             try:
-                instance = model.model_validate(row)
+                # Merge fixed values into the row before validation to account for these.
+                row_with_fixed = {**row, **fixed_values} if fixed_values else row
+                
+                instance = model.model_validate(row_with_fixed)
                 data = instance.model_dump(by_alias=False)
+                
                 empty_required_fields = self.get_empty_required_fields(
                     data, required_fields
                 )
@@ -82,3 +89,87 @@ def get_empty_required_fields(
             if data.get(field) is None
             or (isinstance(data.get(field), str) and not data.get(field).strip())
         ]
+
+    def validate_fixed_values(
+        self, fixed_values: dict, remappings: dict
+    ) -> tuple[bool, list[str]]:
+
+        if not fixed_values:
+            logger.info("No fixed values provided")
+            return True, []
+
+        errors = []
+
+        errors.extend(self.check_for_protected_fields(fixed_values))
+
+        errors.extend(self.check_for_remapped_field_names(fixed_values, remappings))
+
+        errors.extend(self.check_for_remapping_conflicts(fixed_values, remappings))
+
+        errors.extend(self.check_for_valid_aliases(fixed_values, remappings))
+
+        if errors:
+            logger.error(f"Fixed values validation failed: {errors}")
+            raise BulkUploadMetadataException(errors)
+        else:
+            logger.info("Fixed values validation passed")
+
+        return len(errors) == 0, errors
+
+    def check_for_protected_fields(self, fixed_values: dict) -> list[str]:
+        errors = []
+        protected_in_fixed = set(fixed_values.keys()) & self.PROTECTED_FIELDS
+
+        if protected_in_fixed:
+            protected_list = ", ".join(sorted(protected_in_fixed))
+            errors.append(
+                f"Protected fields cannot have fixed values: {protected_list}. "
+                f"These are critical identifiers that must come from the source data."
+            )
+        return errors
+
+    def check_for_remapped_field_names(
+        self, fixed_values: dict, remappings: dict
+    ) -> list[str]:
+        errors = []
+        remapped_values = set(remappings.values())
+
+        for field_name in fixed_values.keys():
+            if field_name in remapped_values:
+                errors.append(
+                    f"Fixed value field '{field_name}' is a remapped value."
+                    f"Use the original field name instead."
+                )
+        return errors
+
+    def check_for_remapping_conflicts(
+        self, fixed_values: dict, remappings: dict
+    ) -> list[str]:
+        errors = []
+        conflicting_fields = set(fixed_values.keys()) & set(remappings.keys())
+
+        if conflicting_fields:
+            conflicting_list = ", ".join(sorted(conflicting_fields))
+            errors.append(
+                f"Fixed values cannot be applied to remapped fields: {conflicting_list}"
+            )
+        return errors
+
+    def check_for_valid_aliases(
+        self, fixed_values: dict, remappings: dict
+    ) -> list[str]:
+        errors = []
+        remapped_values = set(remappings.values())
+
+        for field_name in fixed_values.keys():
+            if field_name not in remapped_values and field_name not in self.model_aliases:
+                errors.append(
+                    f"Fixed value field '{field_name}' is not a valid metadata field alias. "
+                    f"Valid aliases are: {', '.join(sorted(self.model_aliases))}"
+                )
+        return errors
+
+
+
+
+

lambdas/tests/unit/services/test_bulk_upload_metadata_processor_service.py

@@ -377,7 +377,7 @@ def test_duplicates_csv_to_sqs_metadata(mocker, bulk_upload_service):
     mocker.patch.object(
         bulk_upload_service.metadata_mapping_validator_service,
         "validate_and_normalize_metadata",
-        side_effect=lambda records, source_name: (records, [], []),
+        side_effect=lambda records, fixed_values, remappings: (records, [], []),
     )
 
     actual = bulk_upload_service.csv_to_sqs_metadata("fake/path.csv")
@@ -725,7 +725,7 @@ def test_csv_to_sqs_metadata_happy_path(mocker, bulk_upload_service, mock_csv_co
     mocker.patch.object(
         bulk_upload_service.metadata_mapping_validator_service,
         "validate_and_normalize_metadata",
-        side_effect=lambda records, src: (records, [], []),
+        side_effect=lambda records, fixed_values, remappings: (records, [], []),
     )
 
     mock_process_metadata_row = mocker.patch.object(
@@ -784,7 +784,7 @@ def test_csv_to_sqs_metadata_groups_patients_correctly(mocker, bulk_upload_servi
     mocker.patch.object(
         bulk_upload_service.metadata_mapping_validator_service,
         "validate_and_normalize_metadata",
-        side_effect=lambda records, src: (records, [], []),
+        side_effect=lambda records, fixed_values, remappings: (records, [], []),
     )
 
     mocker.patch.object(
@@ -956,7 +956,6 @@ def test_no_remapping_logic(
         )
     ]
 
-
 @freeze_time("2025-02-03T10:00:00")
 def test_validate_expedite_file_happy_path_returns_expected_tuple(test_service):
     ods_code = "A12345"
@@ -1162,3 +1161,100 @@ def test_check_file_status_logs_issue_when_not_clean(mocker, test_service, caplo
         f"Found an issue with the file {file_key}." in record.msg
         for record in caplog.records
     )
+def test_apply_fixed_values_no_fixed_values(test_service, base_metadata_file):
+    result = test_service.apply_fixed_values(base_metadata_file)
+    
+    assert result == base_metadata_file
+
+
+def test_apply_fixed_values_single_field(mocker, base_metadata_file):
+    service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=MockMetadataPreprocessorService(
+            practice_directory="test_practice_directory"
+        ),
+        metadata_heading_remap={},
+        fixed_values={"SECTION": "AR"},
+    )
+    mocker.patch.object(service, "s3_service")
+    
+    result = service.apply_fixed_values(base_metadata_file)
+    
+    assert result.section == "AR"
+    assert result.nhs_number == base_metadata_file.nhs_number
+    assert result.gp_practice_code == base_metadata_file.gp_practice_code
+
+
+def test_apply_fixed_values_multiple_fields(mocker, base_metadata_file):
+    service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=MockMetadataPreprocessorService(
+            practice_directory="test_practice_directory"
+        ),
+        metadata_heading_remap={},
+        fixed_values={
+            "SECTION": "AR",
+            "SUB-SECTION": "Mental Health",
+            "SCAN-ID": "FIXED_SCAN_ID",
+        },
+    )
+    mocker.patch.object(service, "s3_service")
+    
+    result = service.apply_fixed_values(base_metadata_file)
+    
+    assert result.section == "AR"
+    assert result.sub_section == "Mental Health"
+    assert result.scan_id == "FIXED_SCAN_ID"
+    assert result.nhs_number == base_metadata_file.nhs_number
+
+
+def test_apply_fixed_values_overwrites_existing_value(mocker, base_metadata_file):
+    original_section = base_metadata_file.section
+    assert original_section == "LG"
+    
+    service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=MockMetadataPreprocessorService(
+            practice_directory="test_practice_directory"
+        ),
+        metadata_heading_remap={},
+        fixed_values={"SECTION": "AR"},
+    )
+    mocker.patch.object(service, "s3_service")
+    
+    result = service.apply_fixed_values(base_metadata_file)
+    
+    assert result.section == "AR"
+    assert result.section != original_section
+
+
+def test_apply_fixed_values_logs_applied_values(mocker, base_metadata_file, caplog):
+    service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=MockMetadataPreprocessorService(
+            practice_directory="test_practice_directory"
+        ),
+        metadata_heading_remap={},
+        fixed_values={"SECTION": "AR", "SCAN-ID": "TEST_ID"},
+    )
+    mocker.patch.object(service, "s3_service")
+    
+    service.apply_fixed_values(base_metadata_file)
+    
+    log_messages = [record.message for record in caplog.records]
+    assert any("Applied fixed value for field 'SECTION': 'AR'" in msg for msg in log_messages)
+    assert any("Applied fixed value for field 'SCAN-ID': 'TEST_ID'" in msg for msg in log_messages)
+
+
+def test_apply_fixed_values_returns_valid_metadata_file(mocker, base_metadata_file):
+    service = BulkUploadMetadataProcessorService(
+        metadata_formatter_service=MockMetadataPreprocessorService(
+            practice_directory="test_practice_directory"
+        ),
+        metadata_heading_remap={},
+        fixed_values={"SECTION": "AR"},
+    )
+    mocker.patch.object(service, "s3_service")
+    
+    result = service.apply_fixed_values(base_metadata_file)
+    
+    assert isinstance(result, MetadataFile)
+    # Ensure it can be validated again
+    validated = MetadataFile.model_validate(result.model_dump(by_alias=True))
+    assert validated.section == "AR"