Set default settings to boto3 config when connecting to db

Author: robertnovac1

e2e/test_delta_immunization.py

@@ -1,12 +1,9 @@
 from datetime import datetime
 from utils.base_test import ImmunizationBaseTest
 from utils.immunisation_api import parse_location
-from utils.resource import generate_imms_resource
-from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource
+from utils.resource import generate_imms_resource, get_dynamodb_table
 import time
 import copy
-import boto3
-from botocore.config import Config
 import os
 
 
@@ -17,14 +14,9 @@ class TestDeltaImmunization(ImmunizationBaseTest):
     DELETE_OPERATION = "DELETE"
     DPS_SOURCE = "DPS"
 
-    def get_delta_table(self):
-        config = Config(connect_timeout=1, read_timeout=1, retries={"max_attempts": 1})
-        db: DynamoDBServiceResource = boto3.resource("dynamodb", region_name="eu-west-2", config=config)
-        return db.Table(os.getenv("IMMS_DELTA_TABLE_NAME"))
-
     def test_create_delta_imms(self):
         """Should create,update,delete FHIR Immunization resource causing those resources to be stored in Delta table"""
-        imms_delta_table = self.get_delta_table()
+        imms_delta_table = get_dynamodb_table(os.getenv("IMMS_DELTA_TABLE_NAME"))
         # Given
         start_timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
         # Creating two Imms Event, one of them will be updated and second of them will be deleted afterwards

e2e/utils/resource.py

@@ -132,22 +132,21 @@ def get_patient_postal_code(imms: dict):
 
 
 def get_full_row_from_identifier(identifier: str) -> dict:
-    table = get_dynamodb_table()
-
+    table = get_dynamodb_table(os.getenv("DYNAMODB_TABLE_NAME"))
     return table.get_item(Key={"PK": f"Immunization#{identifier}"}).get("Item")
 
 
-def get_dynamodb_table() -> Table:
-    config = Config(connect_timeout=5, read_timeout=5, retries={"max_attempts": 5})
+def get_dynamodb_table(table_name: str) -> Table:
+    config = Config(connect_timeout=60, read_timeout=60, retries={"max_attempts": 3})
     db: DynamoDBServiceResource = boto3.resource("dynamodb", region_name="eu-west-2", config=config)
-    return db.Table(os.getenv("DYNAMODB_TABLE_NAME"))
+    return db.Table(table_name)
 
 
 def delete_imms_records(identifiers: list[str]) -> None:
     """Batch delete immunization records from the DynamoDB table.
     Handles logging internally and does not return anything.
     """
-    table = get_dynamodb_table()
+    table = get_dynamodb_table(os.getenv("DYNAMODB_TABLE_NAME"))
     success_count = 0
     failure_count = 0
     total = len(identifiers)