refactor: Optimize shared API usage plan handling (aws#1973)

* fix: use instance variables for generating shared api usage plan

* add extra log statements

* fix: Added SAR Support Check (aws#1972)

* Added SAR Support Check

* Added docstring and Removed Instance Initialization for Class Method

* set log level explicitly

* update pyyaml version to get the security update (aws#1974)

* fix: use instance variables for generating shared api usage plan

* add extra log statements

* set log level explicitly

* black formatting

* black formatting

Co-authored-by: Cosh_ <[email protected]>
Co-authored-by: Mohamed Elasmar <[email protected]>

Author: 3 people

samtranslator/model/api/api_generator.py

@@ -1,3 +1,4 @@
+import logging
 from collections import namedtuple
 from six import string_types
 from samtranslator.model.intrinsics import ref, fnGetAtt
@@ -24,6 +25,9 @@
 from samtranslator.translator.arn_generator import ArnGenerator
 from samtranslator.model.tags.resource_tagging import get_tag_list
 
+LOG = logging.getLogger(__name__)
+LOG.setLevel(logging.INFO)
+
 _CORS_WILDCARD = "'*'"
 CorsProperties = namedtuple(
     "_CorsProperties", ["AllowMethods", "AllowHeaders", "AllowOrigin", "MaxAge", "AllowCredentials"]
@@ -52,12 +56,20 @@
 GatewayResponseProperties = ["ResponseParameters", "ResponseTemplates", "StatusCode"]
 
 
-class ApiGenerator(object):
-    usage_plan_shared = False
-    stage_keys_shared = list()
-    api_stages_shared = list()
-    depends_on_shared = list()
+class SharedApiUsagePlan(object):
+    """
+    Collects API information from different API resources in the same template,
+    so that these information can be used in the shared usage plan
+    """
+
+    def __init__(self):
+        self.usage_plan_shared = False
+        self.stage_keys_shared = list()
+        self.api_stages_shared = list()
+        self.depends_on_shared = list()
 
+
+class ApiGenerator(object):
     def __init__(
         self,
         logical_id,
@@ -69,6 +81,7 @@ def __init__(
         definition_uri,
         name,
         stage_name,
+        shared_api_usage_plan,
         tags=None,
         endpoint_configuration=None,
         method_settings=None,
@@ -134,6 +147,7 @@ def __init__(
         self.models = models
         self.domain = domain
         self.description = description
+        self.shared_api_usage_plan = shared_api_usage_plan
 
     def _construct_rest_api(self):
         """Constructs and returns the ApiGateway RestApi.
@@ -630,18 +644,19 @@ def _construct_usage_plan(self, rest_api_stage=None):
 
         # create a usage plan for all the Apis
         elif create_usage_plan == "SHARED":
+            LOG.info("Creating SHARED usage plan for all the Apis")
             usage_plan_logical_id = "ServerlessUsagePlan"
-            if self.logical_id not in ApiGenerator.depends_on_shared:
-                ApiGenerator.depends_on_shared.append(self.logical_id)
+            if self.logical_id not in self.shared_api_usage_plan.depends_on_shared:
+                self.shared_api_usage_plan.depends_on_shared.append(self.logical_id)
             usage_plan = ApiGatewayUsagePlan(
-                logical_id=usage_plan_logical_id, depends_on=ApiGenerator.depends_on_shared
+                logical_id=usage_plan_logical_id, depends_on=self.shared_api_usage_plan.depends_on_shared
             )
             api_stage = dict()
             api_stage["ApiId"] = ref(self.logical_id)
             api_stage["Stage"] = ref(rest_api_stage.logical_id)
-            if api_stage not in ApiGenerator.api_stages_shared:
-                ApiGenerator.api_stages_shared.append(api_stage)
-            usage_plan.ApiStages = ApiGenerator.api_stages_shared
+            if api_stage not in self.shared_api_usage_plan.api_stages_shared:
+                self.shared_api_usage_plan.api_stages_shared.append(api_stage)
+            usage_plan.ApiStages = self.shared_api_usage_plan.api_stages_shared
 
             api_key = self._construct_api_key(usage_plan_logical_id, create_usage_plan, rest_api_stage)
             usage_plan_key = self._construct_usage_plan_key(usage_plan_logical_id, create_usage_plan, api_key)
@@ -667,15 +682,16 @@ def _construct_api_key(self, usage_plan_logical_id, create_usage_plan, rest_api_
         """
         if create_usage_plan == "SHARED":
             # create an api key resource for all the apis
+            LOG.info("Creating api key resource for all the Apis from SHARED usage plan")
             api_key_logical_id = "ServerlessApiKey"
             api_key = ApiGatewayApiKey(logical_id=api_key_logical_id, depends_on=[usage_plan_logical_id])
             api_key.Enabled = True
             stage_key = dict()
             stage_key["RestApiId"] = ref(self.logical_id)
             stage_key["StageName"] = ref(rest_api_stage.logical_id)
-            if stage_key not in ApiGenerator.stage_keys_shared:
-                ApiGenerator.stage_keys_shared.append(stage_key)
-            api_key.StageKeys = ApiGenerator.stage_keys_shared
+            if stage_key not in self.shared_api_usage_plan.stage_keys_shared:
+                self.shared_api_usage_plan.stage_keys_shared.append(stage_key)
+            api_key.StageKeys = self.shared_api_usage_plan.stage_keys_shared
         # for create_usage_plan = "PER_API"
         else:
             # create an api key resource for this api

samtranslator/model/sam_resources.py

@@ -857,6 +857,7 @@ def to_cloudformation(self, **kwargs):
         self.Domain = intrinsics_resolver.resolve_parameter_refs(self.Domain)
         self.Auth = intrinsics_resolver.resolve_parameter_refs(self.Auth)
         redeploy_restapi_parameters = kwargs.get("redeploy_restapi_parameters")
+        shared_api_usage_plan = kwargs.get("shared_api_usage_plan")
 
         api_generator = ApiGenerator(
             self.logical_id,
@@ -868,6 +869,7 @@ def to_cloudformation(self, **kwargs):
             self.DefinitionUri,
             self.Name,
             self.StageName,
+            shared_api_usage_plan,
             tags=self.Tags,
             endpoint_configuration=self.EndpointConfiguration,
             method_settings=self.MethodSettings,

samtranslator/translator/translator.py

@@ -6,6 +6,7 @@
     FeatureToggleDefaultConfigProvider,
 )
 from samtranslator.model import ResourceTypeResolver, sam_resources
+from samtranslator.model.api.api_generator import SharedApiUsagePlan
 from samtranslator.translator.verify_logical_id import verify_unique_logical_id
 from samtranslator.model.preferences.deployment_preference_collection import DeploymentPreferenceCollection
 from samtranslator.model.exceptions import (
@@ -111,6 +112,7 @@ def translate(self, sam_template, parameter_values, feature_toggle=None):
         )
         deployment_preference_collection = DeploymentPreferenceCollection()
         supported_resource_refs = SupportedResourceReferences()
+        shared_api_usage_plan = SharedApiUsagePlan()
         document_errors = []
         changed_logical_ids = {}
         for logical_id, resource_dict in self._get_resources_to_iterate(sam_template, macro_resolver):
@@ -130,6 +132,7 @@ def translate(self, sam_template, parameter_values, feature_toggle=None):
                     resource_dict, intrinsics_resolver
                 )
                 kwargs["redeploy_restapi_parameters"] = self.redeploy_restapi_parameters
+                kwargs["shared_api_usage_plan"] = shared_api_usage_plan
                 translated = macro.to_cloudformation(**kwargs)
 
                 supported_resource_refs = macro.get_resource_references(translated, supported_resource_refs)