Python - EC2: Audit results implemented (Pt. 2) (awsdocs#6789)

* This update applies coding standards to the source code and test code.

---------

Co-authored-by: David Souther <[email protected]>

python/cross_service/resilient_service/.gitignore

@@ -1 +1,3 @@
-*.pem
+doc-example-resilience-key-pair-*
+test-doc-example-resilience-key-pair-*
+

python/cross_service/resilient_service/test/conftest.py → python/cross_service/resilient_service/conftest.py

@@ -4,21 +4,16 @@
 """
 Contains common test fixtures used to run unit tests.
 """
-
 from datetime import datetime
-import sys
+
 import boto3
 import pytest
 
-from auto_scaler import AutoScaler
-from load_balancer import LoadBalancer
+import runner
+from auto_scaler import AutoScalingWrapper
+from load_balancer import ElasticLoadBalancerWrapper
 from parameters import ParameterHelper
 from recommendation_service import RecommendationService
-import runner
-
-# This is needed so Python can find test_tools on the path.
-sys.path.append("../..")
-from test_tools.fixtures.common import *
 
 
 class ScenarioData:
@@ -72,7 +67,7 @@ def __init__(self, auto_scaling, elb, ddb, ec2, ssm, iam):
         self.scenario = runner.Runner(
             self.test_resource_path,
             RecommendationService(self.table_name, self.ddb.client),
-            AutoScaler(
+            AutoScalingWrapper(
                 self.resource_prefix,
                 self.inst_type,
                 self.ami_param,
@@ -81,7 +76,7 @@ def __init__(self, auto_scaling, elb, ddb, ec2, ssm, iam):
                 self.ssm.client,
                 self.iam.client,
             ),
-            LoadBalancer(self.tg_name, self.lb_name, self.elb.client),
+            ElasticLoadBalancerWrapper(self.elb.client),
             ParameterHelper(self.table_name, self.ssm.client),
         )
 

python/cross_service/resilient_service/parameters.py

@@ -9,10 +9,6 @@
 log = logging.getLogger(__name__)
 
 
-class ParameterHelperError(Exception):
-    pass
-
-
 # snippet-start:[python.example_code.workflow.ResilientService_ParameterHelper]
 class ParameterHelper:
     """
@@ -21,25 +17,22 @@ class ParameterHelper:
     how the service responds to a health check.
     """
 
-    table = "doc-example-resilient-architecture-table"
-    failure_response = "doc-example-resilient-architecture-failure-response"
-    health_check = "doc-example-resilient-architecture-health-check"
+    table: str = "doc-example-resilient-architecture-table"
+    failure_response: str = "doc-example-resilient-architecture-failure-response"
+    health_check: str = "doc-example-resilient-architecture-health-check"
 
-    def __init__(self, table_name, ssm_client):
+    def __init__(self, table_name: str, ssm_client: boto3.client):
         """
+        Initializes the ParameterHelper class with the necessary parameters.
+
         :param table_name: The name of the DynamoDB table that is used as a recommendation
                            service.
         :param ssm_client: A Boto3 Systems Manager client.
         """
         self.ssm_client = ssm_client
         self.table_name = table_name
 
-    @classmethod
-    def from_client(cls, table_name):
-        ssm_client = boto3.client("ssm")
-        return cls(table_name, ssm_client)
-
-    def reset(self):
+    def reset(self) -> None:
         """
         Resets the Systems Manager parameters to starting values for the demo.
         These are the name of the DynamoDB recommendation table, no response when a
@@ -49,22 +42,34 @@ def reset(self):
         self.put(self.failure_response, "none")
         self.put(self.health_check, "shallow")
 
-    def put(self, name, value):
+    def put(self, name: str, value: str) -> None:
         """
         Sets the value of a named Systems Manager parameter.
 
         :param name: The name of the parameter.
         :param value: The new value of the parameter.
+        :raises ParameterHelperError: If the parameter value cannot be set.
         """
         try:
             self.ssm_client.put_parameter(
                 Name=name, Value=value, Overwrite=True, Type="String"
             )
-            log.info("Setting demo parameter %s to '%s'.", name, value)
+            log.info("Setting parameter %s to '%s'.", name, value)
         except ClientError as err:
-            raise ParameterHelperError(
-                f"Couldn't set parameter {name} to {value}: {err}"
-            )
+            error_code = err.response["Error"]["Code"]
+            log.error(f"Failed to set parameter {name}.")
+            if error_code == "ParameterLimitExceeded":
+                log.error(
+                    "The parameter limit has been exceeded. "
+                    "Consider deleting unused parameters or request a limit increase."
+                )
+            elif error_code == "ParameterAlreadyExists":
+                log.error(
+                    "The parameter already exists and overwrite is set to False. "
+                    "Use Overwrite=True to update the parameter."
+                )
+            log.error(f"Full error:\n\t{err}")
+            pass
 
 
 # snippet-end:[python.example_code.workflow.ResilientService_ParameterHelper]

python/cross_service/resilient_service/recommendation_service.py

@@ -3,6 +3,7 @@
 
 import json
 import logging
+from typing import Any, Dict
 
 import boto3
 from botocore.exceptions import ClientError
@@ -11,7 +12,17 @@
 
 
 class RecommendationServiceError(Exception):
-    def __init__(self, table_name, message):
+    """
+    Custom exception for errors related to the RecommendationService.
+    """
+
+    def __init__(self, table_name: str, message: str):
+        """
+        Initializes the RecommendationServiceError.
+
+        :param table_name: The name of the DynamoDB table where the error occurred.
+        :param message: The error message.
+        """
         self.table_name = table_name
         self.message = message
         super().__init__(self.message)
@@ -24,32 +35,25 @@ class RecommendationService:
     and songs.
     """
 
-    def __init__(self, table_name, dynamodb_client):
+    def __init__(self, table_name: str, dynamodb_client: boto3.client):
         """
+        Initializes the RecommendationService class with the necessary parameters.
+
         :param table_name: The name of the DynamoDB recommendations table.
         :param dynamodb_client: A Boto3 DynamoDB client.
         """
         self.table_name = table_name
         self.dynamodb_client = dynamodb_client
 
-    @classmethod
-    def from_client(cls, table_name):
+    def create(self) -> Dict[str, Any]:
         """
-        Creates this class from a Boto3 client.
-
-        :param table_name: The name of the DynamoDB recommendations table.
-        """
-        ddb_client = boto3.client("dynamodb")
-        return cls(table_name, ddb_client)
-
-    def create(self):
-        """
-        Creates a DynamoDB table to use a recommendation service. The table has a
+        Creates a DynamoDB table to use as a recommendation service. The table has a
         hash key named 'MediaType' that defines the type of media recommended, such as
         Book or Movie, and a range key named 'ItemId' that, combined with the MediaType,
         forms a unique identifier for the recommended item.
 
         :return: Data about the newly created table.
+        :raises RecommendationServiceError: If the table creation fails.
         """
         try:
             response = self.dynamodb_client.create_table(
@@ -70,19 +74,20 @@ def create(self):
             log.info("Table %s created.", self.table_name)
         except ClientError as err:
             if err.response["Error"]["Code"] == "ResourceInUseException":
-                log.info("Table %s exists, nothing to be do.", self.table_name)
+                log.info("Table %s exists, nothing to be done.", self.table_name)
             else:
                 raise RecommendationServiceError(
                     self.table_name, f"ClientError when creating table: {err}."
                 )
         else:
             return response
 
-    def populate(self, data_file):
+    def populate(self, data_file: str) -> None:
         """
         Populates the recommendations table from a JSON file.
 
         :param data_file: The path to the data file.
+        :raises RecommendationServiceError: If the table population fails.
         """
         try:
             with open(data_file) as data:
@@ -97,9 +102,11 @@ def populate(self, data_file):
                 self.table_name, f"Couldn't populate table from {data_file}: {err}"
             )
 
-    def destroy(self):
+    def destroy(self) -> None:
         """
         Deletes the recommendations table.
+
+        :raises RecommendationServiceError: If the table deletion fails.
         """
         try:
             self.dynamodb_client.delete_table(TableName=self.table_name)

python/cross_service/resilient_service/requirements.txt

@@ -1,3 +1,4 @@
 boto3>=1.26.79
 pytest>=7.2.1
-requests>=2.29.0
+requests>=2.29.0
+coloredlogs>=15.0.1