Merge pull request #5 from MacHu-GWU/1.0.1

1.0.1

.github/workflows/main.yml

@@ -15,13 +15,13 @@ jobs:
     name: "${{ matrix.os }} Python ${{ matrix.python-version }}"
     runs-on: "${{ matrix.os }}" # for all available VM runtime, see this: https://docs.github.com/en/free-pro-team@latest/actions/reference/specifications-for-github-hosted-runners
     env: # define environment variables
-      USING_COVERAGE: "3.6,3.7,3.8,3.9"
+      USING_COVERAGE: "3.7,3.8,3.9,3.10"
     strategy:
       matrix:
         os: ["ubuntu-latest", "windows-latest"]
 #        os: ["ubuntu-latest", ] # for debug only
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10"]
-#        python-version: ["3.6", ] # for debug only
+        python-version: ["3.7", "3.8", "3.9", "3.10"]
+#        python-version: ["3.7", ] # for debug only
         exclude:
           - os: windows-latest # this is a useless exclude rules for demonstration use only
             python-version: 2.7

cottonformation/__init__.py

@@ -32,21 +32,43 @@
 
 
 try:
-    from .core import constant, helpers, exc
-    from .core.constant import DeletionPolicy, UpdateReplacePolicy
+    from .core import (
+        constant,
+        helpers,
+        exc,
+    )
+    from .core.constant import DeletionPolicyEnum, UpdateReplacePolicyEnum
     from .core.model import (
         # data model
-        Parameter, Property, Resource, Output, Export,
-        Rule, Mapping, Condition, Transform,
-        ResourceGroup, Tag,
-
+        Parameter,
+        Property,
+        Resource,
+        Output,
+        Export,
+        Rule,
+        Mapping,
+        Condition,
+        Transform,
+        ResourceGroup,
+        Tag,
         # intrinsic function
-        Ref, Base64, Cidr, FindInMap, GetAtt, GetAZs,
-        ImportValue, Join, Select, Split, Sub,
-
+        Ref,
+        Base64,
+        Cidr,
+        FindInMap,
+        GetAtt,
+        GetAZs,
+        ImportValue,
+        Join,
+        Select,
+        Split,
+        Sub,
         # condition function
-        Equals, If, Not, And, Or,
-
+        Equals,
+        If,
+        Not,
+        And,
+        Or,
         # pseudo parameter
         AWS_ACCOUNT_ID,
         AWS_NOTIFICATION_ARNS,
@@ -61,5 +83,10 @@
     from .core.template import Template
     from .core.stack import Stack
     from .core.env import Env
-except ImportError as e:
+except ImportError as e:  # pragma: no cover
+    pass
+
+try:
+    from aws_cloudformation.stack import Parameter as ParameterValue
+except ImportError as e:  # pragma: no cover
     pass

cottonformation/_version.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.8"
+__version__ = "1.0.1"
 
 if __name__ == "__main__":  # pragma: no cover
     print(__version__)

cottonformation/code/__init__.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 """
-AWS CloudFormation resource specification is a machine readable json file
+AWS CloudFormation resource specification is a machine-readable json file
 that includes detailed information about AWS Resource, Property specification.
 
 The AWS Resource class declaration code is auto generated base on this file.

cottonformation/code/spec.py

@@ -89,17 +89,21 @@ class User:
 def _find_type_hint_and_validator(
     prop: typing.Union['Property', 'ResourceProperty'],
     prop_type_class_name: str,
-    cycled_class_name_set: typing.Set[str]
+    cycled_class_name_set: typing.Set[str],
+    spec_data: dict,
 ) -> typing.Tuple[
-    str, typing.Union[str, None], typing.Union[str, None]]:
+    str,
+    typing.Union[str, None],
+    typing.Union[str, None]
+]:
     """
     We need to define type hint and validators for attrs.
 
-    Those syntax is generally based on the type, primitive_type, item_type,
+    The syntax is generally based on the type, primitive_type, item_type,
     primitive_item_type of the property. This method implements the logic to
     generate type hint and validator python code.
 
-    **I know the logic is complicate, and there's a lots of edge case need to be
+    **I know the logic is complicated, and there's a lots of edge case need to be
     handled properly. This is the best I can do now**.
     """
     # --- Debug ---
@@ -190,6 +194,18 @@ def _find_type_hint_and_validator(
     #         converter = "{}.from_dict".format("Prop" + parent_class_name)
 
     elif bool(prop.Type):
+        # special handler for appflow, some property are special type but not defined
+        if isinstance(prop, Property):
+            expected_property_types_key = "{}.{}".format(
+                prop.PropertyTypesKey.split(".")[0],
+                prop.Name,
+            )
+            if expected_property_types_key not in prop.SpecData["PropertyTypes"]:
+                type_hint = "typing.Optional[dict]"
+                validator = "attr.validators.optional(attr.validators.instance_of(dict))"
+                converter = None
+                return type_hint, validator, converter
+
         parent_class_name = prop.ResourceName + prop.Type
         type_hint = "typing.Union['{}', dict]".format("Prop" + parent_class_name)
         if (parent_class_name == prop_type_class_name) or (
@@ -232,6 +248,7 @@ def _get_type_hint(
     prop: typing.Union['Property', 'ResourceProperty'],
     prop_type_class_name: str,
     cycled_class_name_set: typing.Set[str],
+    spec_data: dict,
 ) -> str:
     """
     For example::
@@ -241,13 +258,19 @@ class User:
             # name: {{ type_hint }}
             name: str
     """
-    return _find_type_hint_and_validator(prop, prop_type_class_name, cycled_class_name_set)[0]
+    return _find_type_hint_and_validator(
+        prop,
+        prop_type_class_name,
+        cycled_class_name_set,
+        spec_data,
+    )[0]
 
 
 def _get_vali_def(
     prop: typing.Union['Property', 'ResourceProperty'],
     prop_type_class_name: str,
     cycled_class_name_set: typing.Set[str],
+    spec_data: dict,
 ) -> str:
     """
     For example::
@@ -257,12 +280,20 @@ class User:
             # name: str = attr.ib(validators={{ vali_def }})
             name: str = attr.ib(validators=attr.validators.instance_of(str))
     """
-    return _find_type_hint_and_validator(prop, prop_type_class_name, cycled_class_name_set)[1]
+    return _find_type_hint_and_validator(
+        prop,
+        prop_type_class_name,
+        cycled_class_name_set,
+        spec_data,
+    )[1]
 
 
-def _get_converter(prop: typing.Union['Property', 'ResourceProperty'],
-                   prop_type_class_name: str,
-                   cycled_class_name_set: typing.Set[str]) -> typing.Union[str, None]:
+def _get_converter(
+    prop: typing.Union['Property', 'ResourceProperty'],
+    prop_type_class_name: str,
+    cycled_class_name_set: typing.Set[str],
+    spec_data: dict,
+) -> typing.Union[str, None]:
     """
     For example::
 
@@ -271,7 +302,12 @@ class User:
             # name: str = attr.ib(validators={{ vali_def }})
             name: str = attr.ib(validators=attr.validators.instance_of(str))
     """
-    return _find_type_hint_and_validator(prop, prop_type_class_name, cycled_class_name_set)[2]
+    return _find_type_hint_and_validator(
+        prop,
+        prop_type_class_name,
+        cycled_class_name_set,
+        spec_data,
+    )[2]
 
 
 def _make_template(filename):
@@ -289,12 +325,17 @@ def _make_template(filename):
 
 @attr.s
 class Property:
+    """
+    Represent a ``.PropertyTypes.${property_name}.Properties.${prop_name}`` object
+    in cft-spec.json file.
+    """
     Name: str = attr.field(default=None)
     SystemName: str = attr.field(default=None)
     ServiceName: str = attr.field(default=None)
     ResourceName: str = attr.field(default=None)
     PropertyName: str = attr.field(default=None)
     PropertyFullName: str = attr.field(default=None)
+    PropertyTypesKey: str = attr.field(default=None)
 
     Required: bool = attr.field(default=None)
     Type: str = attr.field(default=None)
@@ -308,6 +349,8 @@ class Property:
     PropertyTypeClassName: str = attr.field(default=None)
     CycledClassNameSet: typing.Set[str] = attr.field(default=None)
 
+    SpecData: dict = attr.field(default=None)
+
     @property
     def sort_key(self):
         return _get_sort_key(self)
@@ -322,19 +365,23 @@ def class_name(self):
 
     @property
     def type_hint(self):
-        return _get_type_hint(self, self.PropertyTypeClassName, self.CycledClassNameSet)
+        return _get_type_hint(self, self.PropertyTypeClassName, self.CycledClassNameSet, self.SpecData)
 
     @property
     def vali_def(self):
-        return _get_vali_def(self, self.PropertyTypeClassName, self.CycledClassNameSet)
+        return _get_vali_def(self, self.PropertyTypeClassName, self.CycledClassNameSet, self.SpecData)
 
     @property
     def converter(self):
-        return _get_converter(self, self.PropertyTypeClassName, self.CycledClassNameSet)
+        return _get_converter(self, self.PropertyTypeClassName, self.CycledClassNameSet, self.SpecData)
 
 
 @attr.s
 class PropertyType:
+    """
+    Represent a ``.PropertyTypes.${property_name}`` object
+    in cft-spec.json file.
+    """
     Name: str = attr.field(default=None)
     SystemName: str = attr.field(default=None)
     ServiceName: str = attr.field(default=None)
@@ -351,6 +398,8 @@ class PropertyType:
 
     Properties: typing.List[Property] = attr.field(default=None)
 
+    SpecData: dict = attr.field(default=None)
+
     @property
     def class_name(self):
         return "Prop" + self.PropertyFullName.replace(".", "")
@@ -367,6 +416,10 @@ def find_property(self, name):
 
 @attr.s
 class ResourceProperty:
+    """
+    Represent a ``.ResourceTypes.${resource_name}.Properties.${property_name}`` object
+    in cft-spec.json file.
+    """
     Name: str = attr.field(default=None)
     SystemName: str = attr.field(default=None)
     ServiceName: str = attr.field(default=None)
@@ -381,6 +434,9 @@ class ResourceProperty:
     Documentation: str = attr.field(default=None)
     UpdateType: str = attr.field(default=None)
     DuplicatesAllowed: str = attr.field(default=None)
+    Data: dict = attr.field(default=None)
+
+    SpecData: dict = attr.field(default=None)
 
     @property
     def sort_key(self):
@@ -392,19 +448,23 @@ def attr_name(self):
 
     @property
     def type_hint(self):
-        return _get_type_hint(self, "NEVER_HAPPEN", set())
+        return _get_type_hint(self, "NEVER_HAPPEN", set(), self.SpecData)
 
     @property
     def vali_def(self):
-        return _get_vali_def(self, "NEVER_HAPPEN", set())
+        return _get_vali_def(self, "NEVER_HAPPEN", set(), self.SpecData)
 
     @property
     def converter(self):
-        return _get_converter(self, "NEVER_HAPPEN", set())
+        return _get_converter(self, "NEVER_HAPPEN", set(), self.SpecData)
 
 
 @attr.s
 class ResourceAttribute:
+    """
+    Represent a ``.ResourceTypes.${resource_name}.Attributes.${attribute_name}`` object
+    in cft-spec.json file.
+    """
     Name: str = attr.ib(default=None)
     SystemName: str = attr.field(default=None)
     ServiceName: str = attr.field(default=None)
@@ -425,6 +485,8 @@ def method_name(self):
 @attr.s
 class ResourceType:
     """
+    Represent a ``.ResourceTypes.${resource_name}`` object
+    in cft-spec.json file.
     """
     Name: str = attr.field(default=None)
     SystemName: str = attr.field(default=None)
@@ -471,7 +533,9 @@ def new(cls) -> 'CftSpec':
                 property_dct["PropertyName"] = property_name
                 property_dct["PropertyFullName"] = ResourceName + "." + property_name
                 property_dct["PropertyTypeClassName"] = ResourceName + PropertyName
-                property = Property(**property_dct)
+                property_dct["PropertyTypesKey"] = property_type_name
+
+                property = Property(SpecData=spec_data, **property_dct)
                 property_properties.append(property)
 
                 # --- Debug ---
@@ -493,7 +557,8 @@ def new(cls) -> 'CftSpec':
             property_type_dct["PropertyName"] = PropertyName
             property_type_dct["PropertyFullName"] = PropertyFullName
             property_type_dct["Properties"] = property_properties
-            property_type = PropertyType(**property_type_dct)
+
+            property_type = PropertyType(SpecData=spec_data, **property_type_dct)
 
             property_type_list.append(property_type)
 
@@ -505,13 +570,14 @@ def new(cls) -> 'CftSpec':
             resource_properties = list()
             for resource_property_name, resource_property_dct in resource_type_dct.get("Properties", {}).items():
                 # print(f"--- {resource_property_name} ---")
+                resource_property_dct["Data"] = resource_property_dct.copy()
                 resource_property_dct["Name"] = resource_property_name
                 resource_property_dct["SystemName"] = SystemName
                 resource_property_dct["ServiceName"] = ServiceName
                 resource_property_dct["ResourceName"] = ResourceName
                 resource_property_dct["PropertyName"] = resource_property_name
                 resource_property_dct["PropertyFullName"] = ResourceName + "." + resource_property_name
-                resource_property = ResourceProperty(**resource_property_dct)
+                resource_property = ResourceProperty(SpecData=spec_data, **resource_property_dct)
                 resource_properties.append(resource_property)
 
                 # sometime, both Type and PrimitiveType exists, in this case

cottonformation/code/spec_file/__init__.py

@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+

cottonformation/code/spec_file/downloader.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+
+import json
+import requests
+from .paths import path_cft_spec_json
+
+
+def download_spec_file(skip_if_exists: bool = True):
+    """
+    The spec file download link can be found at
+    https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-resource-specification.html
+
+    By default, it overwrite the file
+    """
+    if path_cft_spec_json.exists():
+        if skip_if_exists:
+            return
+    url = "https://d1uauaxba7bl26.cloudfront.net/latest/gzip/CloudFormationResourceSpecification.json"
+    path_cft_spec_json.write_text(requests.get(url).text, encoding="utf-8")
+
+
+def read_spec_file() -> dict:
+    """
+    Read data of the spec file.
+    """
+    return json.loads(path_cft_spec_json.read_text(encoding="utf-8"))