Enable pyupgrade lint rules and autofix errors. (#1246)

* Enable pyupgrade lint rules and autofix errors.

* Use appropriate exception type.

* Update ruff.toml

Author: mfjarvis

lambdas/ack_backend/src/update_ack_file.py

@@ -4,7 +4,7 @@
 import os
 import time
 from copy import deepcopy
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from io import BytesIO, StringIO
 
 from botocore.exceptions import ClientError
@@ -40,7 +40,7 @@
 
 
 def _generated_date() -> str:
-    return datetime.now(timezone.utc).isoformat()[:-13] + ".000Z"
+    return datetime.now(UTC).isoformat()[:-13] + ".000Z"
 
 
 def _make_ack_data_dict_identifier_information(

lambdas/ack_backend/tests/utils/utils_for_ack_backend_tests.py

@@ -2,7 +2,6 @@
 
 import json
 from copy import deepcopy
-from typing import Optional
 
 from boto3 import client as boto3_client
 
@@ -12,7 +11,7 @@
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
 
 
-def add_audit_entry_to_table(dynamodb_client, batch_event_message_id: str, record_count: Optional[int] = None) -> None:
+def add_audit_entry_to_table(dynamodb_client, batch_event_message_id: str, record_count: int | None = None) -> None:
     """Add an entry to the audit table"""
     audit_table_entry = {"status": {"S": "Preprocessed"}, "message_id": {"S": batch_event_message_id}}
 

lambdas/backend/src/controller/aws_apig_event_utils.py

@@ -1,7 +1,5 @@
 """Utility module for interacting with the AWS API Gateway event provided to controllers"""
 
-from typing import Optional
-
 from aws_lambda_typing.events import APIGatewayProxyEventV1
 
 from controller.constants import E_TAG_HEADER_NAME, SUPPLIER_SYSTEM_HEADER_NAME
@@ -19,7 +17,7 @@ def get_path_parameter(event: APIGatewayProxyEventV1, param_name: str) -> str:
 
 def get_supplier_system_header(event: APIGatewayProxyEventV1) -> str:
     """Retrieves the supplier system header from the API Gateway event. Raises an Unauthorized error if not present."""
-    supplier_system: Optional[str] = dict_utils.get_field(dict(event), "headers", SUPPLIER_SYSTEM_HEADER_NAME)
+    supplier_system: str | None = dict_utils.get_field(dict(event), "headers", SUPPLIER_SYSTEM_HEADER_NAME)
 
     if supplier_system is None:
         # SupplierSystem header must be provided for looking up permissions
@@ -31,7 +29,7 @@ def get_supplier_system_header(event: APIGatewayProxyEventV1) -> str:
 def get_resource_version_header(event: APIGatewayProxyEventV1) -> str:
     """Retrieves the resource version header from the API Gateway event. Raises a ResourceVersionNotProvided if not
     present."""
-    resource_version_header: Optional[str] = dict_utils.get_field(dict(event), "headers", E_TAG_HEADER_NAME)
+    resource_version_header: str | None = dict_utils.get_field(dict(event), "headers", E_TAG_HEADER_NAME)
 
     if resource_version_header is None:
         raise ResourceVersionNotProvidedError(resource_type="Immunization")

lambdas/backend/src/controller/aws_apig_response_utils.py

@@ -1,10 +1,9 @@
 """Utility module providing helper functions for dealing with response formats for AWS API Gateway"""
 
 import json
-from typing import Optional
 
 
-def create_response(status_code: int, body: Optional[dict | str] = None, headers: Optional[dict] = None) -> dict:
+def create_response(status_code: int, body: dict | str | None = None, headers: dict | None = None) -> dict:
     """Creates response body as per Lambda -> API Gateway proxy integration"""
     if body is not None:
         if isinstance(body, dict):

lambdas/backend/src/controller/fhir_api_exception_handler.py

@@ -2,7 +2,7 @@
 
 import functools
 import uuid
-from typing import Callable, Type
+from collections.abc import Callable
 
 from common.clients import logger
 from common.models.errors import (
@@ -31,7 +31,7 @@
     create_operation_outcome,
 )
 
-_CUSTOM_EXCEPTION_TO_STATUS_MAP: dict[Type[Exception], int] = {
+_CUSTOM_EXCEPTION_TO_STATUS_MAP: dict[type[Exception], int] = {
     InconsistentResourceVersionError: 400,
     InconsistentIdentifierError: 400,  # Identifier refers to the local FHIR identifier composed of system and value.
     InconsistentIdError: 400,  # ID refers to the top-level ID of the FHIR resource.

lambdas/backend/src/controller/parameter_parser.py

@@ -1,6 +1,5 @@
 import datetime
 from dataclasses import dataclass, field
-from typing import Optional
 
 from common.models.constants import RedisHashKeys
 from common.models.utils.generic_utils import nhs_number_mod11_check
@@ -25,9 +24,9 @@
 class SearchParams:
     patient_identifier: str
     immunization_targets: set[str]
-    date_from: Optional[datetime.date]
-    date_to: Optional[datetime.date]
-    include: Optional[str]
+    date_from: datetime.date | None
+    date_to: datetime.date | None
+    include: str | None
 
     def __repr__(self):
         return str(self.__dict__)
@@ -111,7 +110,7 @@ def process_mandatory_params(params: dict[str, list[str]]) -> tuple[str, list[st
 
 def process_optional_params(
     params: dict[str, list[str]],
-) -> tuple[Optional[datetime.date], Optional[datetime.date], Optional[str]]:
+) -> tuple[datetime.date | None, datetime.date | None, str | None]:
     """Parse optional params (date.from, date.to, _include).
     Raises ParameterExceptionError for any validation error.
     """
@@ -205,7 +204,7 @@ def check_elements_valid(elements: list[str]) -> bool:
     return set(elements).issubset({IdentifierSearchElement.ID, IdentifierSearchElement.META})
 
 
-def validate_and_retrieve_identifier_search_params(params: dict[str, list[str]]) -> tuple[str, Optional[set[str]]]:
+def validate_and_retrieve_identifier_search_params(params: dict[str, list[str]]) -> tuple[str, set[str] | None]:
     contains_no_patient_vacc_type_params = check_identifier_search_params_contain_no_incorrect_keys(params)
 
     if not contains_no_patient_vacc_type_params:

lambdas/backend/src/local_lambda.py

@@ -1,3 +1,3 @@
 def load_string(file_path: str):
-    with open(file_path, "r", encoding="utf-8") as f:
+    with open(file_path, encoding="utf-8") as f:
         return f.read()

lambdas/backend/src/log_structure.py

@@ -33,8 +33,8 @@ def _get_operation_outcome(result) -> dict:
     operation_outcome = {}
     status = "500"
     status_code = "Exception"
-    diagnostics = str()
-    record = str()
+    diagnostics = ""
+    record = ""
     if isinstance(result, dict):
         status = str(result["statusCode"])
         status_code = "Completed successfully"

lambdas/backend/src/repository/fhir_repository.py

@@ -1,7 +1,6 @@
 import os
 import time
 from dataclasses import dataclass
-from typing import Optional
 
 import boto3
 import botocore.exceptions
@@ -96,7 +95,7 @@ def __init__(self, table: Table):
 
     def get_immunization_by_identifier(
         self, identifier: Identifier
-    ) -> tuple[Optional[dict], Optional[ImmunizationRecordMetadata]]:
+    ) -> tuple[dict | None, ImmunizationRecordMetadata | None]:
         response = self.table.query(
             IndexName="IdentifierGSI",
             KeyConditionExpression=Key("IdentifierPK").eq(self._make_identifier_pk(identifier)),
@@ -120,7 +119,7 @@ def get_immunization_by_identifier(
 
     def get_immunization_resource_and_metadata_by_id(
         self, imms_id: str, include_deleted: bool = False
-    ) -> tuple[Optional[dict], Optional[ImmunizationRecordMetadata]]:
+    ) -> tuple[dict | None, ImmunizationRecordMetadata | None]:
         """Retrieves the immunization resource and metadata from the VEDS table"""
         response = self.table.get_item(Key={"PK": _make_immunization_pk(imms_id)})
         item = response.get("Item")

lambdas/backend/src/service/fhir_service.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import uuid
-from typing import Any, Optional, cast
+from typing import Any, cast
 from uuid import uuid4
 
 from fhir.resources.R4B.bundle import (
@@ -73,7 +73,7 @@ def __init__(
         self.validator = validator
 
     def get_immunization_by_identifier(
-        self, identifier: Identifier, supplier_name: str, elements: Optional[set[str]]
+        self, identifier: Identifier, supplier_name: str, elements: set[str] | None
     ) -> FhirBundle:
         """
         Get an Immunization by its ID. Returns a FHIR Bundle containing the search results.
@@ -195,10 +195,10 @@ def search_immunizations(
         nhs_number: str,
         vaccine_types: set[str],
         supplier_system: str,
-        date_from: Optional[datetime.date],
-        date_to: Optional[datetime.date],
-        include: Optional[str],
-        invalid_immunization_targets: Optional[list[str]] = None,
+        date_from: datetime.date | None,
+        date_to: datetime.date | None,
+        include: str | None,
+        invalid_immunization_targets: list[str] | None = None,
     ) -> FhirBundle:
         """
         Finds all instances of Immunization(s) for a specified patient for the given specified vaccine type(s).
@@ -301,9 +301,9 @@ def search_immunizations(
     def _filter_search_results_by_date_and_status(
         self,
         immunizations: list[dict],
-        date_from: Optional[datetime.date],
-        date_to: Optional[datetime.date],
-        status: Optional[str],
+        date_from: datetime.date | None,
+        date_to: datetime.date | None,
+        status: str | None,
     ) -> list[dict]:
         return [
             immunization
@@ -313,7 +313,7 @@ def _filter_search_results_by_date_and_status(
             and validate_has_status(immunization, status)
         ]
 
-    def is_valid_date_from(self, immunization: dict, date_from: Optional[datetime.date]):
+    def is_valid_date_from(self, immunization: dict, date_from: datetime.date | None):
         """
         Returns False if immunization occurrence is earlier than the date_from, or True otherwise
         (also returns True if date_from is None)
@@ -327,7 +327,7 @@ def is_valid_date_from(self, immunization: dict, date_from: Optional[datetime.da
 
         return occurrence_datetime.date() >= date_from
 
-    def is_valid_date_to(self, immunization: dict, date_to: Optional[datetime.date]):
+    def is_valid_date_to(self, immunization: dict, date_to: datetime.date | None):
         """
         Returns False if immunization occurrence is later than the date_to, or True otherwise
         (also returns True if date_to is None)
@@ -343,9 +343,9 @@ def is_valid_date_to(self, immunization: dict, date_to: Optional[datetime.date])
 
     @staticmethod
     def make_identifier_search_bundle(
-        resource: Optional[dict],
-        version_id: Optional[int],
-        elements: Optional[set[str]],
+        resource: dict | None,
+        version_id: int | None,
+        elements: set[str] | None,
         identifier: Identifier,
         base_url: str,
     ) -> FhirBundle: