Refactor code

Author: GavinZZ

requirements/base.txt

@@ -2,5 +2,5 @@ boto3>=1.19.5,==1.*
 jsonschema<5,>=3.2  # TODO: evaluate risk of removing jsonschema 3.x support
 typing_extensions>=4.4,<5 # 3.7 doesn't have Literal
 
-# resource validation & schema generation, requiring features in >=1.10
-pydantic~=1.10
+# resource validation & schema generation
+pydantic~=1.8.0

samtranslator/internal/schema_source/common.py

@@ -56,6 +56,15 @@ class BaseModel(LenientBaseModel):
     class Config:
         extra = Extra.forbid
 
+    def __getattribute__(self, __name: str) -> Any:
+        """Overloading get attribute operation"""
+        attr_value = super().__getattribute__(__name)
+        if isinstance(attr_value, PassThroughProp):
+            # Access __root__ attribute to get actual value from PassThroughProp
+            # See https://github.com/aws/serverless-application-model/blob/develop/samtranslator/internal/schema_source/common.py#L19
+            return attr_value.__root__
+        return attr_value
+
 
 # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/intrinsic-function-reference-ref.html
 class Ref(BaseModel):

samtranslator/validator/resource_validator.py

@@ -1,47 +1,18 @@
 """A resource validator to help validate resource properties and raise exception when some value is unexpected."""
-from typing import Any, Dict, Type
+from typing import Any, Dict, Type, TypeVar
 
 from pydantic import BaseModel
 
-
-class Model:
-    """
-    Wrapper class around a SAM schema BaseModel to with a new functional "get" method
-    """
-
-    def __init__(self, model: BaseModel) -> None:
-        self.model = model
-
-    def _process_attr_value(self, attr_value: Any) -> Any:
-        if isinstance(attr_value, BaseModel):
-            if "__root__" in attr_value.__dict__:
-                return attr_value.__dict__["__root__"]
-            return Model(attr_value)
-        return attr_value
-
-    def _get_item(self, attr_value: Any) -> Any:
-        if isinstance(attr_value, list):
-            return [self._process_attr_value(attr) for attr in attr_value]
-        return self._process_attr_value(attr_value)
-
-    def get(self, attr_key: str, default_value: Any = None) -> Any:
-        """Return the value for key if key is in Model properties else default."""
-        attr_value = self.model.__dict__.get(attr_key, default_value)
-        return self._get_item(attr_value)
-
-    def __getitem__(self, attr_key: str) -> Any:
-        """Return the value for key if key is in Model properties else raise KeyError exception."""
-        attr_value = self.model.__dict__[attr_key]
-        return self._get_item(attr_value)
+T = TypeVar("T", bound=BaseModel)
 
 
 # Note: For compabitliy issue, we should ONLY use this with new abstraction/resources.
-def to_model(resource_properties: Dict[Any, Any], cls: Type[BaseModel]) -> Model:
+def to_model(resource_properties: Dict[Any, Any], cls: Type[T]) -> T:
     """
     Given properties of a SAM resource return a typed object from the definitions of SAM schema model
 
     param:
         resource_properties: properties from input template
         cls: SAM schema models
     """
-    return Model(cls(**resource_properties))
+    return cls.parse_obj(resource_properties)

tests/validator/test_resource_validator.py

@@ -4,9 +4,9 @@
 
 from pydantic import BaseModel
 from pydantic.error_wrappers import ValidationError
+from samtranslator.internal.schema_source.aws_serverless_connector import Properties as ConnectorProperties
 from samtranslator.validator.resource_validator import to_model
 from samtranslator.yaml_helper import yaml_parse
-from schema_source.aws_serverless_connector import Properties as ConnectorProperties
 
 BASE_PATH = os.path.dirname(__file__)
 CONNECTOR_INPUT_FOLDER = os.path.join(BASE_PATH, "input", "connector")
@@ -65,32 +65,24 @@ def test_connector_model_get_operation(self):
             self.connector_template,
             ConnectorProperties,
         )
-        self.assertEqual(connector_model.get("Source").get("Arn"), "random-arn")
-        self.assertEqual(connector_model.get("Source").get("Type"), "random-type")
-        self.assertEqual(connector_model.get("Source").get("Id"), None)
-        self.assertEqual(connector_model.get("Destination").get("Id"), "MyTable")
-        self.assertEqual(connector_model.get("Permissions"), ["Read"])
-        self.assertEqual(connector_model.get("FakeProperty"), None)
-
-        self.assertEqual(connector_model["Source"]["Arn"], "random-arn")
-        self.assertEqual(connector_model["Source"]["Type"], "random-type")
-        self.assertEqual(connector_model["Destination"]["Id"], "MyTable")
-        self.assertEqual(connector_model["Permissions"], ["Read"])
+        self.assertEqual(connector_model.Source.Arn, "random-arn")
+        self.assertEqual(connector_model.Source.Type, "random-type")
+        self.assertEqual(connector_model.Source.Id, None)
+        self.assertEqual(connector_model.Destination.Id, "MyTable")
+        self.assertEqual(connector_model.Permissions, ["Read"])
 
     def test_model_get_operation(self):
         model = to_model(self.model_template, ValidatiorBaseModel)
-        self.assertEqual(model["Properties"]["Key"], {"A": {"value": 10}})
-        self.assertEqual(model["Properties"]["Key"]["A"], {"value": 10})
-        self.assertEqual(model["Properties"]["Hello"], ["1", "2", "3"])
-        self.assertEqual(model.get("Properties").get("Random")["value"], 5)
-        self.assertEqual(model.get("DoNotExist"), None)
-
-        self.assertEqual(len(model["Contents"]), 3)
-        self.assertEqual(model["Contents"][0]["Content"]["Tags"]["A"], "hello")
-        self.assertEqual(model["Contents"][0]["Content"]["Tags"]["B"], 5)
-        self.assertEqual(model["Contents"][1]["Content"]["Tags"]["A"], "wow")
-        self.assertEqual(model["Contents"][0]["Content"]["Tags"].get("C"), None)
-        self.assertEqual(model["Contents"][2]["Content"]["Tags"]["B"], -5)
+        self.assertEqual(model.Properties.Key, {"A": {"value": 10}})
+        self.assertEqual(model.Properties.Key["A"], {"value": 10})
+        self.assertEqual(model.Properties.Hello, ["1", "2", "3"])
+        self.assertEqual(model.Properties.Random.value, 5)
+
+        self.assertEqual(len(model.Contents), 3)
+        self.assertEqual(model.Contents[0].Content.Tags.A, "hello")
+        self.assertEqual(model.Contents[0].Content.Tags.B, 5)
+        self.assertEqual(model.Contents[1].Content.Tags.A, "wow")
+        self.assertEqual(model.Contents[2].Content.Tags.B, -5)
 
 
 class TestModelValidatorFailure(TestCase):