Resolve sonar warnings

Author: dlzhry2nhs

lambdas/recordprocessor/src/constants.py

@@ -4,7 +4,7 @@
 from enum import StrEnum
 
 # Once Python projects are moved to /lambdas consider consolidating constants common to batch in
-# /shared/src/common/constants/batch_constants.py
+# /shared/src/common/constants/batch_constants.py (VED-881)
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
 ACK_BUCKET_NAME = os.getenv("ACK_BUCKET_NAME")
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")

lambdas/recordprocessor/src/file_level_validation.py

@@ -117,7 +117,7 @@ def file_level_validation(incoming_message_body: dict) -> dict:
             "csv_dict_reader": csv_reader,
         }
 
-    except (InvalidHeaders, NoOperationPermissions, Exception) as error:
+    except Exception as error:
         logger.error("Error in file_level_validation: %s", error)
 
         # NOTE: The Exception may occur before the file_id, file_key and created_at_formatted_string are assigned

lambdas/recordprocessor/src/logging_decorator.py

@@ -39,11 +39,17 @@ def wrapper(*args, **kwargs):
             generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data)
             return result
 
-        except (InvalidHeaders, NoOperationPermissions, Exception) as e:
-            message = (
-                str(e) if (isinstance(e, InvalidHeaders) or isinstance(e, NoOperationPermissions)) else "Server error"
-            )
-            status_code = 400 if isinstance(e, InvalidHeaders) else 403 if isinstance(e, NoOperationPermissions) else 500
+        except Exception as e:
+            if isinstance(e, InvalidHeaders):
+                message = str(e)
+                status_code = 400
+            elif isinstance(e, NoOperationPermissions):
+                message = str(e)
+                status_code = 403
+            else:
+                message = "Server error"
+                status_code = 500
+
             additional_log_data = {
                 "statusCode": status_code,
                 "message": message,

lambdas/recordprocessor/src/models/utils.py

@@ -1,7 +1,6 @@
 import uuid
 from dataclasses import dataclass
 from enum import Enum
-from typing import Union
 
 
 class Severity(str, Enum):
@@ -25,7 +24,7 @@ class Code(str, Enum):
 class UnhandledResponseError(RuntimeError):
     """Use this error when the response from an external service (ex: dynamodb) can't be handled"""
 
-    response: Union[dict, str]
+    response: dict | str
     message: str
 
     def __str__(self):
@@ -53,7 +52,7 @@ class ImmunizationApiError(RuntimeError):
 
     status_code: int
     request: dict
-    response: Union[dict, str]
+    response: dict | str
 
 
 def create_operation_outcome(resource_id: str, severity: Severity, code: Code, diagnostics: str) -> dict: