Resolve some Sonar warnings.

Author: mfjarvis

backend/src/models/obtain_field_value.py

@@ -66,10 +66,10 @@ def patient_gender(imms: dict):
     def patient_address_postal_code(imms: dict):
         """Obtains patient_address_postal_code value"""
         patient = get_contained_patient(imms)
-        contained_patient_postalCode = [x for x in patient.get("address") if len(x.get("postalCode", "")) >= 1][0][
+        contained_patient_postal_code = [x for x in patient.get("address") if len(x.get("postalCode", "")) >= 1][0][
             "postalCode"
         ]
-        return contained_patient_postalCode
+        return contained_patient_postal_code
 
     @staticmethod
     def organization_identifier_value(imms: dict):

backend/src/models/utils/validation_utils.py

@@ -49,7 +49,7 @@ def get_target_disease_codes(immunization: dict):
 
 def convert_disease_codes_to_vaccine_type(
     disease_codes_input: list,
-) -> Union[str, None]:
+) -> str | None:
     """
     Takes a list of disease codes and returns the corresponding vaccine type if found,
     otherwise raises a value error

backend/src/parameter_parser.py

@@ -11,6 +11,8 @@
 from models.errors import ParameterException
 from models.utils.generic_utils import nhs_number_mod11_check
 
+ERROR_MESSAGE_DUPLICATED_PARAMETERS = 'Parameters may not be duplicated. Use commas for "or".'
+
 ParamValue = list[str]
 ParamContainer = dict[str, ParamValue]
 
@@ -161,7 +163,7 @@ def parse_multi_value_query_parameters(
         multi_value_query_params: dict[str, list[str]],
     ) -> ParamContainer:
         if any([len(v) > 1 for k, v in multi_value_query_params.items()]):
-            raise ParameterException('Parameters may not be duplicated. Use commas for "or".')
+            raise ParameterException(ERROR_MESSAGE_DUPLICATED_PARAMETERS)
         params = [(k, split_and_flatten(v)) for k, v in multi_value_query_params.items()]
 
         return dict(params)
@@ -175,7 +177,7 @@ def parse_body_params(aws_event: APIGatewayProxyEventV1) -> ParamContainer:
             parsed_body = parse_qs(decoded_body)
 
             if any([len(v) > 1 for k, v in parsed_body.items()]):
-                raise ParameterException('Parameters may not be duplicated. Use commas for "or".')
+                raise ParameterException(ERROR_MESSAGE_DUPLICATED_PARAMETERS)
             items = dict((k, split_and_flatten(v)) for k, v in parsed_body.items())
             return items
         return {}
@@ -184,7 +186,7 @@ def parse_body_params(aws_event: APIGatewayProxyEventV1) -> ParamContainer:
     body_params = parse_body_params(aws_event)
 
     if len(set(query_params.keys()) & set(body_params.keys())) > 0:
-        raise ParameterException('Parameters may not be duplicated. Use commas for "or".')
+        raise ParameterException(ERROR_MESSAGE_DUPLICATED_PARAMETERS)
 
     parsed_params = {
         key: sorted(query_params.get(key, []) + body_params.get(key, []))

batch_processor_filter/src/batch_processor_filter_service.py

@@ -63,7 +63,7 @@ def apply_filter(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
                 filename,
             )
             raise EventAlreadyProcessingForSupplierAndVaccTypeError(
-                f"Batch event already processing for supplier: " f"{supplier} and vacc type: {vaccine_type}"
+                f"Batch event already processing for supplier: {supplier} and vacc type: {vaccine_type}"
             )
 
         self._batch_audit_repository.update_status(message_id, FileStatus.PROCESSING)

recordprocessor/src/batch_processor.py

@@ -19,7 +19,6 @@
 from send_to_kinesis import send_to_kinesis
 from clients import logger
 from file_level_validation import file_level_validation, file_is_empty, move_file
-from errors import NoOperationPermissions, InvalidHeaders
 from utils_for_recordprocessor import get_csv_content_dict_reader
 from typing import Optional
 
@@ -35,11 +34,7 @@ def process_csv_to_fhir(incoming_message_body: dict) -> int:
     try:
         incoming_message_body["encoder"] = encoder
         interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
-    except (
-        InvalidHeaders,
-        NoOperationPermissions,
-        Exception,
-    ) as e:  # pylint: disable=broad-exception-caught
+    except Exception as e:  # pylint: disable=broad-exception-caught
         logger.error(f"File level validation failed: {e}")  # If the file is invalid, processing should cease
         return 0