Brianwithay21 · Feb 10, 2022 · Sep 30, 2021 · Feb 8, 2022
diff --git a/.pylintrc b/.pylintrc
@@ -1,6 +1,6 @@
 [MASTER]
 
-ignore=CVS,models.py,handlers.py
+ignore=CVS,models.py,handlers.py,hook_models.py,hook_handlers.py,target_model.py
 jobs=1
 persistent=yes
 

diff --git a/python/rpdk/python/__init__.py b/python/rpdk/python/__init__.py
@@ -1,5 +1,5 @@
 import logging
 
-__version__ = "2.1.4"
+__version__ = "2.1.5"
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())
diff --git a/python/rpdk/python/codegen.py b/python/rpdk/python/codegen.py
@@ -15,6 +15,7 @@
 from rpdk.core.init import input_with_validation
 from rpdk.core.jsonutils.resolver import ContainerType, resolve_models
 from rpdk.core.plugin_base import LanguagePlugin
+from rpdk.core.project import ARTIFACT_TYPE_HOOK
 
 from . import __version__
 from .resolver import contains_model, translate_type
@@ -38,7 +39,9 @@ class Python36LanguagePlugin(LanguagePlugin):
     MODULE_NAME = __name__
     NAME = "python36"
     RUNTIME = "python3.6"
-    ENTRY_POINT = "{}.handlers.resource"
+    HOOK_ENTRY_POINT = "{}.handlers.hook"
+    RESOURCE_ENTRY_POINT = "{}.handlers.resource"
+    TEST_ENTRY_POINT = "{}.handlers.test_entrypoint"
     CODE_URI = "build/"
 
     def __init__(self):
@@ -63,6 +66,18 @@ def _init_from_project(self, project):
     def _init_settings(self, project):
         LOG.debug("Writing settings")
 
+        project.runtime = self.RUNTIME
+        if project.artifact_type == ARTIFACT_TYPE_HOOK:
+            project.entrypoint = self.HOOK_ENTRY_POINT.format(self.package_name)
+            project.test_entrypoint = project.entrypoint.replace(
+                ".hook", ".test_entrypoint"
+            )
+        else:
+            project.entrypoint = self.RESOURCE_ENTRY_POINT.format(self.package_name)
+            project.test_entrypoint = project.entrypoint.replace(
+                ".resource", ".test_entrypoint"
+            )
+
         self._use_docker = self._use_docker or input_with_validation(
             "Use docker for platform-independent packaging (Y/n)?\n",
             validate_no,
@@ -79,12 +94,6 @@ def init(self, project):
         self._init_from_project(project)
         self._init_settings(project)
 
-        project.runtime = self.RUNTIME
-        project.entrypoint = self.ENTRY_POINT.format(self.package_name)
-        project.test_entrypoint = project.entrypoint.replace(
-            ".resource", ".test_entrypoint"
-        )
-
         def _render_template(path, **kwargs):
             LOG.debug("Writing '%s'", path)
             template = self.env.get_template(path.name)
@@ -103,11 +112,7 @@ def _copy_resource(path, resource_name=None):
         LOG.debug("Making folder '%s'", handler_package_path)
         handler_package_path.mkdir(parents=True, exist_ok=True)
         _copy_resource(handler_package_path / "__init__.py")
-        _render_template(
-            handler_package_path / "handlers.py",
-            support_lib_pkg=SUPPORT_LIB_PKG,
-            type_name=project.type_name,
-        )
+        self.init_handlers(project, handler_package_path)
         # models.py produced by generate
 
         # project support files
@@ -144,12 +149,25 @@ def _copy_resource(path, resource_name=None):
 
         LOG.debug("Init complete")
 
+    def init_handlers(self, project, handler_package_path):
+        path = handler_package_path / "handlers.py"
+        if project.artifact_type == ARTIFACT_TYPE_HOOK:
+            template = self.env.get_template("hook_handlers.py")
+        else:
+            template = self.env.get_template("handlers.py")
+        contents = template.render(
+            support_lib_pkg=SUPPORT_LIB_PKG, type_name=project.type_name
+        )
+        project.safewrite(path, contents)
+
     def generate(self, project):
         LOG.debug("Generate started")
 
         self._init_from_project(project)
-
-        models = resolve_models(project.schema)
+        if project.artifact_type == ARTIFACT_TYPE_HOOK:
+            models = resolve_models(project.schema, "HookInputModel")
+        else:
+            models = resolve_models(project.schema)
 
         if project.configuration_schema:
             configuration_schema_path = (
@@ -165,7 +183,11 @@ def generate(self, project):
 
         path = self.package_root / self.package_name / "models.py"
         LOG.debug("Writing file: %s", path)
-        template = self.env.get_template("models.py")
+        if project.artifact_type == ARTIFACT_TYPE_HOOK:
+            template = self.env.get_template("hook_models.py")
+        else:
+            template = self.env.get_template("models.py")
+
         contents = template.render(support_lib_pkg=SUPPORT_LIB_PKG, models=models)
         project.overwrite(path, contents)
 

diff --git a/python/rpdk/python/templates/hook_handlers.py b/python/rpdk/python/templates/hook_handlers.py
@@ -0,0 +1,98 @@
+import logging
+from typing import Any, MutableMapping, Optional
+
+from {{support_lib_pkg}} import (
+    BaseHookHandlerRequest,
+    HandlerErrorCode,
+    Hook,
+    HookInvocationPoint,
+    OperationStatus,
+    ProgressEvent,
+    SessionProxy,
+    exceptions,
+)
+
+from .models import HookHandlerRequest, TypeConfigurationModel
+
+# Use this logger to forward log messages to CloudWatch Logs.
+LOG = logging.getLogger(__name__)
+TYPE_NAME = "{{ type_name }}"
+
+hook = Hook(TYPE_NAME, TypeConfigurationModel)
+test_entrypoint = hook.test_entrypoint
+
+
+@hook.handler(HookInvocationPoint.CREATE_PRE_PROVISION)
+def pre_create_handler(
+        session: Optional[SessionProxy],
+        request: HookHandlerRequest,
+        callback_context: MutableMapping[str, Any],
+        type_configuration: TypeConfigurationModel
+) -> ProgressEvent:
+    target_model = request.hookContext.targetModel
+    progress: ProgressEvent = ProgressEvent(
+        status=OperationStatus.IN_PROGRESS
+    )
+    # TODO: put code here
+
+    # Example:
+    try:
+        # Reading the Resource Hook's target properties
+        resource_properties = target_model.get("resourceProperties")
+
+        if isinstance(session, SessionProxy):
+            client = session.client("s3")
+        # Setting Status to success will signal to cfn that the hook operation is complete
+        progress.status = OperationStatus.SUCCESS
+    except TypeError as e:
+        # exceptions module lets CloudFormation know the type of failure that occurred
+        raise exceptions.InternalFailure(f"was not expecting type {e}")
+        # this can also be done by returning a failed progress event
+        # return ProgressEvent.failed(HandlerErrorCode.InternalFailure, f"was not expecting type {e}")
+
+    return progress
+
+
+@hook.handler(HookInvocationPoint.UPDATE_PRE_PROVISION)
+def pre_update_handler(
+        session: Optional[SessionProxy],
+        request: BaseHookHandlerRequest,
+        callback_context: MutableMapping[str, Any],
+        type_configuration: TypeConfigurationModel
+) -> ProgressEvent:
+    target_model = request.hookContext.targetModel
+    progress: ProgressEvent = ProgressEvent(
+        status=OperationStatus.IN_PROGRESS
+    )
+    # TODO: put code here
+
+    # Example:
+    try:
+        # A Hook that does not allow a resource's encryption algorithm to be modified
+
+        # Reading the Resource Hook's target current properties and previous properties
+        resource_properties = target_model.get("resourceProperties")
+        previous_properties = target_model.get("previousResourceProperties")
+
+        if resource_properties.get("encryptionAlgorithm") != previous_properties.get("encryptionAlgorithm"):
+            progress.status = OperationStatus.FAILED
+            progress.message = "Encryption algorithm can not be changed"
+        else:
+            progress.status = OperationStatus.SUCCESS
+    except TypeError as e:
+        progress = ProgressEvent.failed(HandlerErrorCode.InternalFailure, f"was not expecting type {e}")
+
+    return progress
+
+
+@hook.handler(HookInvocationPoint.DELETE_PRE_PROVISION)
+def pre_delete_handler(
+        session: Optional[SessionProxy],
+        request: BaseHookHandlerRequest,
+        callback_context: MutableMapping[str, Any],
+        type_configuration: TypeConfigurationModel
+) -> ProgressEvent:
+    # TODO: put code here
+    return ProgressEvent(
+        status=OperationStatus.SUCCESS
+    )
diff --git a/python/rpdk/python/templates/hook_models.py b/python/rpdk/python/templates/hook_models.py
@@ -0,0 +1,78 @@
+# DO NOT modify this file by hand, changes will be overwritten
+import sys
+from dataclasses import dataclass
+from inspect import getmembers, isclass
+from typing import (
+    AbstractSet,
+    Any,
+    Generic,
+    Mapping,
+    MutableMapping,
+    Optional,
+    Sequence,
+    Type,
+    TypeVar,
+)
+
+from cloudformation_cli_python_lib.interface import BaseHookHandlerRequest, BaseModel
+from cloudformation_cli_python_lib.recast import recast_object
+from cloudformation_cli_python_lib.utils import deserialize_list
+
+T = TypeVar("T")
+
+
+def set_or_none(value: Optional[Sequence[T]]) -> Optional[AbstractSet[T]]:
+    if value:
+        return set(value)
+    return None
+
+
+@dataclass
+class HookHandlerRequest(BaseHookHandlerRequest):
+    pass
+
+
+{% for model, properties in models.items() %}
+@dataclass
+class {{ model }}(BaseModel):
+    {% for name, type in properties.items() %}
+    {{ name }}: Optional[{{ type|translate_type }}]
+    {% endfor %}
+
+    @classmethod
+    def _deserialize(
+        cls: Type["_{{ model }}"],
+        json_data: Optional[Mapping[str, Any]],
+    ) -> Optional["_{{ model }}"]:
+        if not json_data:
+            return None
+        {% if model.endswith("ResourceModel") %}
+        dataclasses = {n: o for n, o in getmembers(sys.modules[__name__]) if isclass(o)}
+        recast_object(cls, json_data, dataclasses)
+        {% endif %}
+        return cls(
+            {% for name, type in properties.items() %}
+            {% set container = type.container %}
+            {% set resolved_type = type.type %}
+            {% if container == ContainerType.MODEL %}
+            {{ name }}={{ resolved_type }}._deserialize(json_data.get("{{ name }}")),
+            {% elif container == ContainerType.SET %}
+            {{ name }}=set_or_none(json_data.get("{{ name }}")),
+            {% elif container == ContainerType.LIST %}
+            {% if type | contains_model %}
+            {{name}}=deserialize_list(json_data.get("{{ name }}"), {{resolved_type.type}}),
+            {% else %}
+            {{ name }}=json_data.get("{{ name }}"),
+            {% endif %}
+            {% else %}
+            {{ name }}=json_data.get("{{ name }}"),
+            {% endif %}
+            {% endfor %}
+        )
+
+
+# work around possible type aliasing issues when variable has same name as a model
+_{{ model }} = {{ model }}
+
+
+{% endfor -%}
diff --git a/python/rpdk/python/templates/requirements.txt b/python/rpdk/python/templates/requirements.txt
@@ -1 +1 @@
-{{ support_lib_name }}>=2.1.3
+{{ support_lib_name }}>=2.1.9
diff --git a/python/rpdk/python/templates/target_model.py b/python/rpdk/python/templates/target_model.py
@@ -0,0 +1,77 @@
+# DO NOT modify this file by hand, changes will be overwritten
+import sys
+from dataclasses import dataclass
+from inspect import getmembers, isclass
+from typing import (
+    AbstractSet,
+    Any,
+    Generic,
+    Mapping,
+    MutableMapping,
+    Optional,
+    Sequence,
+    Type,
+    TypeVar,
+)
+
+from cloudformation_cli_python_lib.interface import BaseModel
+from cloudformation_cli_python_lib.recast import recast_object
+from cloudformation_cli_python_lib.utils import deserialize_list
+
+T = TypeVar("T")
+
+
+def set_or_none(value: Optional[Sequence[T]]) -> Optional[AbstractSet[T]]:
+    if value:
+        return set(value)
+    return None
+
+
+{% for model, properties in models.items() %}
+@dataclass
+class {{ model }}(BaseModel):
+    {% for name, type in properties.items() %}
+    {{ name }}: Optional[{{ type|translate_type }}]
+    {% endfor %}
+
+    @classmethod
+    def _deserialize(
+        cls: Type["_{{ model }}"],
+        json_data: Optional[Mapping[str, Any]],
+    ) -> Optional["_{{ model }}"]:
+        if not json_data:
+            return None
+        {% if model == (target_name) %}
+        data = dict(filter(lambda e: e[0] in cls.__dataclass_fields__, json_data.items()))
+        if not data:
+            return None
+
+        dataclasses = {n: o for n, o in getmembers(sys.modules[__name__]) if isclass(o)}
+        recast_object(cls, data, dataclasses)
+        {% endif %}
+        return cls(
+            {% for name, type in properties.items() %}
+            {% set container = type.container %}
+            {% set resolved_type = type.type %}
+            {% if container == ContainerType.MODEL %}
+            {{ name }}={{ resolved_type }}._deserialize(json_data.get("{{ name }}")),
+            {% elif container == ContainerType.SET %}
+            {{ name }}=set_or_none(json_data.get("{{ name }}")),
+            {% elif container == ContainerType.LIST %}
+            {% if type | contains_model %}
+            {{name}}=deserialize_list(json_data.get("{{ name }}"), {{resolved_type.type}}),
+            {% else %}
+            {{ name }}=json_data.get("{{ name }}"),
+            {% endif %}
+            {% else %}
+            {{ name }}=json_data.get("{{ name }}"),
+            {% endif %}
+            {% endfor %}
+        )
+
+
+# work around possible type aliasing issues when variable has same name as a model
+_{{ model }} = {{ model }}
+
+
+{% endfor -%}
diff --git a/setup.py b/setup.py
@@ -38,7 +38,7 @@ def find_version(*file_paths):
     include_package_data=True,
     zip_safe=True,
     python_requires=">=3.6",
-    install_requires=["cloudformation-cli>=0.2.13", "types-dataclasses>=0.1.5"],
+    install_requires=["cloudformation-cli>=0.2.23", "types-dataclasses>=0.1.5"],
     entry_points={
         "rpdk.v1.languages": [
             "python37 = rpdk.python.codegen:Python37LanguagePlugin",
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		{{ support_lib_name }}>=2.1.3
		{{ support_lib_name }}>=2.1.9