refactor: Refactor ImplicitHttpApiPlugin and ImplicitRestApiPlugin

samtranslator/plugins/api/implicit_http_api_plugin.py

@@ -1,16 +1,14 @@
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional, Type, cast
 
 from samtranslator.model.intrinsics import make_conditional
-from samtranslator.model.naming import GeneratedLogicalId
 from samtranslator.plugins.api.implicit_api_plugin import ImplicitApiPlugin
 from samtranslator.public.open_api import OpenApiEditor
-from samtranslator.public.exceptions import InvalidEventException
 from samtranslator.public.sdk.resource import SamResourceType, SamResource
 from samtranslator.sdk.template import SamTemplate
 from samtranslator.validator.value_validator import sam_expect
 
 
-class ImplicitHttpApiPlugin(ImplicitApiPlugin):
+class ImplicitHttpApiPlugin(ImplicitApiPlugin[Type[OpenApiEditor]]):
     """
     This plugin provides Implicit Http API shorthand syntax in the SAM Spec.
 
@@ -28,26 +26,22 @@ class ImplicitHttpApiPlugin(ImplicitApiPlugin):
                                              in OpenApi. Does **not** configure the API by any means.
     """
 
-    def __init__(self) -> None:
-        """
-        Initializes the plugin
-        """
-        super(ImplicitHttpApiPlugin, self).__init__(ImplicitHttpApiPlugin.__name__)
-
-    def _setup_api_properties(self) -> None:
-        """
-        Sets up properties that are distinct to this plugin
-        """
-        self.implicit_api_logical_id = GeneratedLogicalId.implicit_http_api()
-        self.implicit_api_condition = "ServerlessHttpApiCondition"
-        self.api_event_type = "HttpApi"
-        self.api_type = SamResourceType.HttpApi.value
-        self.api_id_property = "ApiId"
-        self.editor = OpenApiEditor
-
-    def _process_api_events(  # type: ignore[no-untyped-def]
-        self, function, api_events, template, condition=None, deletion_policy=None, update_replace_policy=None
-    ):
+    API_ID_EVENT_PROPERTY = "ApiId"
+    IMPLICIT_API_LOGICAL_ID = "ServerlessHttpApi"
+    IMPLICIT_API_CONDITION = "ServerlessHttpApiCondition"
+    API_EVENT_TYPE = "HttpApi"
+    SERVERLESS_API_RESOURCE_TYPE = SamResourceType.HttpApi.value
+    EDITOR_CLASS = OpenApiEditor
+
+    def _process_api_events(
+        self,
+        function: SamResource,
+        api_events: Dict[str, Dict[str, Any]],
+        template: SamTemplate,
+        condition: Optional[str] = None,
+        deletion_policy: Optional[str] = None,
+        update_replace_policy: Optional[str] = None,
+    ) -> None:
         """
         Actually process given HTTP API events. Iteratively adds the APIs to OpenApi JSON in the respective
         AWS::Serverless::HttpApi resource from the template
@@ -58,21 +52,16 @@ def _process_api_events(  # type: ignore[no-untyped-def]
         :param str condition: optional; this is the condition that is on the function with the API event
         """
 
-        for logicalId, event in api_events.items():
+        for event_id, event in api_events.items():
             # api_events only contains HttpApi events
             event_properties = event.get("Properties", {})
 
-            if not isinstance(event_properties, dict):
-                raise InvalidEventException(
-                    logicalId,
-                    "Event 'Properties' must be an Object. If you're using YAML, this may be an indentation issue.",
-                )
-
+            sam_expect(event_properties, event_id, "", is_sam_event=True).to_be_a_map("Properties should be a map.")
             if not event_properties:
-                event["Properties"] = event_properties
+                event["Properties"] = event_properties  # We are updating its Properties
+
             self._add_implicit_api_id_if_necessary(event_properties)  # type: ignore[no-untyped-call]
 
-            api_id = self._get_api_id(event_properties)  # type: ignore[no-untyped-call]
             path = event_properties.get("Path", "")
             method = event_properties.get("Method", "")
             # If no path and method specified, add the $default path and ANY method
@@ -81,49 +70,20 @@ def _process_api_events(  # type: ignore[no-untyped-def]
                 method = "x-amazon-apigateway-any-method"
                 event_properties["Path"] = path
                 event_properties["Method"] = method
-            elif not path or not method:
-                key = "Path" if not path else "Method"
-                raise InvalidEventException(logicalId, "Event is missing key '{}'.".format(key))
-
-            if not isinstance(path, str) or not isinstance(method, str):
-                key = "Path" if not isinstance(path, str) else "Method"
-                raise InvalidEventException(logicalId, "Api Event must have a String specified for '{}'.".format(key))
 
-            # !Ref is resolved by this time. If it is not a string, we can't parse/use this Api.
-            if api_id and not isinstance(api_id, str):
-                raise InvalidEventException(
-                    logicalId, "Api Event's ApiId must be a string referencing an Api in the same template."
-                )
+            api_id, path, method = self._validate_api_event(event_id, event_properties)
+            self._update_resource_attributes_from_api_event(
+                api_id, path, method, condition, deletion_policy, update_replace_policy
+            )
 
-            api_dict_condition = self.api_conditions.setdefault(api_id, {})
-            method_conditions = api_dict_condition.setdefault(path, {})
-            method_conditions[method] = condition
-
-            api_dict_deletion = self.api_deletion_policies.setdefault(api_id, set())
-            api_dict_deletion.add(deletion_policy)
-
-            api_dict_update_replace = self.api_update_replace_policies.setdefault(api_id, set())
-            api_dict_update_replace.add(update_replace_policy)
-
-            self._add_api_to_swagger(logicalId, event_properties, template)  # type: ignore[no-untyped-call]
+            self._add_api_to_swagger(event_id, event_properties, template)  # type: ignore[no-untyped-call]
             if "RouteSettings" in event_properties:
-                self._add_route_settings_to_api(logicalId, event_properties, template, condition)
-            api_events[logicalId] = event
+                self._add_route_settings_to_api(event_id, event_properties, template, condition)
+            api_events[event_id] = event
 
         # We could have made changes to the Events structure. Write it back to function
         function.properties["Events"].update(api_events)
 
-    def _add_implicit_api_id_if_necessary(self, event_properties):  # type: ignore[no-untyped-def]
-        """
-        Events for implicit APIs will *not* have the RestApiId property. Absence of this property means this event
-        is associated with the AWS::Serverless::Api ImplicitAPI resource.
-        This method solidifies this assumption by adding RestApiId property to events that don't have them.
-
-        :param dict event_properties: Dictionary of event properties
-        """
-        if "ApiId" not in event_properties:
-            event_properties["ApiId"] = {"Ref": self.implicit_api_logical_id}
-
     def _generate_implicit_api_resource(self) -> Dict[str, Any]:
         """
         Uses the implicit API in this file to generate an Implicit API resource
@@ -136,12 +96,6 @@ def _get_api_definition_from_editor(self, editor: OpenApiEditor) -> Dict[str, An
         """
         return editor.openapi
 
-    def _get_api_resource_type_name(self) -> str:
-        """
-        Returns the type of API resource
-        """
-        return "AWS::Serverless::HttpApi"
-
     def _add_route_settings_to_api(
         self, event_id: str, event_properties: Dict[str, Any], template: SamTemplate, condition: Optional[str]
     ) -> None:
@@ -155,7 +109,7 @@ def _add_route_settings_to_api(
         :param string condition: Condition on this HttpApi event (if any)
         """
 
-        api_id = self._get_api_id(event_properties)  # type: ignore[no-untyped-call]
+        api_id = self._get_api_id(event_properties)
         resource = cast(SamResource, template.get(api_id))  # TODO: make this not an assumption
 
         path = event_properties["Path"]

samtranslator/plugins/api/implicit_rest_api_plugin.py

@@ -1,13 +1,13 @@
-from typing import Any, Dict
+from typing import Any, Dict, Optional, Type
 
-from samtranslator.model.naming import GeneratedLogicalId
 from samtranslator.plugins.api.implicit_api_plugin import ImplicitApiPlugin
 from samtranslator.public.swagger import SwaggerEditor
-from samtranslator.public.exceptions import InvalidEventException
 from samtranslator.public.sdk.resource import SamResourceType, SamResource
+from samtranslator.sdk.template import SamTemplate
+from samtranslator.validator.value_validator import sam_expect
 
 
-class ImplicitRestApiPlugin(ImplicitApiPlugin):
+class ImplicitRestApiPlugin(ImplicitApiPlugin[Type[SwaggerEditor]]):
     """
     This plugin provides Implicit API shorthand syntax in the SAM Spec.
     https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api
@@ -26,29 +26,24 @@ class ImplicitRestApiPlugin(ImplicitApiPlugin):
 
     * API Event Source (In Core Translator): ONLY adds the Lambda Integration ARN to appropriate method/path
                                              in Swagger. Does **not** configure the API by any means.
-
     """
 
-    def __init__(self) -> None:
-        """
-        Initialize the plugin
-        """
-        super(ImplicitRestApiPlugin, self).__init__(ImplicitRestApiPlugin.__name__)
-
-    def _setup_api_properties(self) -> None:
-        """
-        Sets up properties that are distinct to this plugin
-        """
-        self.implicit_api_logical_id = GeneratedLogicalId.implicit_api()
-        self.implicit_api_condition = "ServerlessRestApiCondition"
-        self.api_event_type = "Api"
-        self.api_type = SamResourceType.Api.value
-        self.api_id_property = "RestApiId"
-        self.editor = SwaggerEditor
-
-    def _process_api_events(  # type: ignore[no-untyped-def]
-        self, function, api_events, template, condition=None, deletion_policy=None, update_replace_policy=None
-    ):
+    API_ID_EVENT_PROPERTY = "RestApiId"
+    IMPLICIT_API_LOGICAL_ID = "ServerlessRestApi"
+    IMPLICIT_API_CONDITION = "ServerlessRestApiCondition"
+    API_EVENT_TYPE = "Api"
+    SERVERLESS_API_RESOURCE_TYPE = SamResourceType.Api.value
+    EDITOR_CLASS = SwaggerEditor
+
+    def _process_api_events(
+        self,
+        function: SamResource,
+        api_events: Dict[str, Dict[str, Any]],
+        template: SamTemplate,
+        condition: Optional[str] = None,
+        deletion_policy: Optional[str] = None,
+        update_replace_policy: Optional[str] = None,
+    ) -> None:
         """
         Actually process given API events. Iteratively adds the APIs to Swagger JSON in the respective Serverless::Api
         resource from the template
@@ -59,84 +54,40 @@ def _process_api_events(  # type: ignore[no-untyped-def]
         :param str condition: optional; this is the condition that is on the function with the API event
         """
 
-        for logicalId, event in api_events.items():
+        for event_id, event in api_events.items():
 
             event_properties = event.get("Properties", {})
             if not event_properties:
                 continue
 
-            if not isinstance(event_properties, dict):
-                raise InvalidEventException(
-                    logicalId,
-                    "Event 'Properties' must be an Object. If you're using YAML, this may be an indentation issue.",
-                )
+            sam_expect(event_properties, event_id, "", is_sam_event=True).to_be_a_map("Properties should be a map.")
 
             self._add_implicit_api_id_if_necessary(event_properties)  # type: ignore[no-untyped-call]
 
-            api_id = self._get_api_id(event_properties)  # type: ignore[no-untyped-call]
-            try:
-                path = event_properties["Path"]
-                method = event_properties["Method"]
-            except KeyError as e:
-                raise InvalidEventException(logicalId, "Event is missing key {}.".format(e))
-
-            if not isinstance(path, str):
-                raise InvalidEventException(logicalId, "Api Event must have a String specified for 'Path'.")
-            if not isinstance(method, str):
-                raise InvalidEventException(logicalId, "Api Event must have a String specified for 'Method'.")
-
-            # !Ref is resolved by this time. If it is not a string, we can't parse/use this Api.
-            if api_id and not isinstance(api_id, str):
-                raise InvalidEventException(
-                    logicalId, "Api Event's RestApiId must be a string referencing an Api in the same template."
-                )
-
-            api_dict_condition = self.api_conditions.setdefault(api_id, {})
-            method_conditions = api_dict_condition.setdefault(path, {})
-            method_conditions[method] = condition
+            api_id, path, method = self._validate_api_event(event_id, event_properties)
+            self._update_resource_attributes_from_api_event(
+                api_id, path, method, condition, deletion_policy, update_replace_policy
+            )
 
-            api_dict_deletion = self.api_deletion_policies.setdefault(api_id, set())
-            api_dict_deletion.add(deletion_policy)
+            self._add_api_to_swagger(event_id, event_properties, template)  # type: ignore[no-untyped-call]
 
-            api_dict_update_replace = self.api_update_replace_policies.setdefault(api_id, set())
-            api_dict_update_replace.add(update_replace_policy)
-
-            self._add_api_to_swagger(logicalId, event_properties, template)  # type: ignore[no-untyped-call]
-
-            api_events[logicalId] = event
+            api_events[event_id] = event
 
         # We could have made changes to the Events structure. Write it back to function
         function.properties["Events"].update(api_events)
 
-    def _add_implicit_api_id_if_necessary(self, event_properties):  # type: ignore[no-untyped-def]
-        """
-        Events for implicit APIs will *not* have the RestApiId property. Absence of this property means this event
-        is associated with the Serverless::Api ImplicitAPI resource. This method solifies this assumption by adding
-        RestApiId property to events that don't have them.
-
-        :param dict event_properties: Dictionary of event properties
-        """
-        if "RestApiId" not in event_properties:
-            event_properties["RestApiId"] = {"Ref": self.implicit_api_logical_id}
-
     def _generate_implicit_api_resource(self) -> Dict[str, Any]:
         """
         Uses the implicit API in this file to generate an Implicit API resource
         """
         return ImplicitApiResource().to_dict()
 
-    def _get_api_definition_from_editor(self, editor):  # type: ignore[no-untyped-def]
+    def _get_api_definition_from_editor(self, editor: SwaggerEditor) -> Dict[str, Any]:
         """
         Helper function to return the OAS definition from the editor
         """
         return editor.swagger
 
-    def _get_api_resource_type_name(self) -> str:
-        """
-        Returns the type of API resource
-        """
-        return "AWS::Serverless::Api"
-
 
 class ImplicitApiResource(SamResource):
     """

samtranslator/validator/value_validator.py

@@ -73,24 +73,40 @@ def to_not_be_none(self, message: Optional[str] = "") -> T:
     # alias methods:
     #
     def to_be_a_map(self, message: Optional[str] = "") -> Dict[str, Any]:
+        """
+        Return the value with type hint "Dict[str, Any]".
+        Raise InvalidResourceException/InvalidEventException if the value is not.
+        """
         return cast(Dict[str, Any], self.to_be_a(ExpectedType.MAP, message))
 
     def to_be_a_list(self, message: Optional[str] = "") -> T:
         return self.to_be_a(ExpectedType.LIST, message)
 
     def to_be_a_list_of(self, expected_type: ExpectedType, message: Optional[str] = "") -> T:
+        """
+        Return the value with type hint "List[T]".
+        Raise InvalidResourceException/InvalidEventException if the value is not.
+        """
         value = self.to_be_a(ExpectedType.LIST, message)
         for index, item in enumerate(value):  # type: ignore
             sam_expect(
                 item, self.resource_id, f"{self.property_identifier}[{index}]", is_sam_event=self.is_sam_event
             ).to_be_a(expected_type, message)
         return value
 
-    def to_be_a_string(self, message: Optional[str] = "") -> T:
-        return self.to_be_a(ExpectedType.STRING, message)
+    def to_be_a_string(self, message: Optional[str] = "") -> str:
+        """
+        Return the value with type hint "str".
+        Raise InvalidResourceException/InvalidEventException if the value is not.
+        """
+        return cast(str, self.to_be_a(ExpectedType.STRING, message))
 
-    def to_be_an_integer(self, message: Optional[str] = "") -> T:
-        return self.to_be_a(ExpectedType.INTEGER, message)
+    def to_be_an_integer(self, message: Optional[str] = "") -> int:
+        """
+        Return the value with type hint "int".
+        Raise InvalidResourceException/InvalidEventException if the value is not.
+        """
+        return cast(int, self.to_be_a(ExpectedType.INTEGER, message))
 
 
 sam_expect = _ResourcePropertyValueValidator