Generate models for hook targets

python/rpdk/python/codegen.py

@@ -9,7 +9,12 @@
 from rpdk.core.data_loaders import resource_stream
 from rpdk.core.exceptions import DownstreamError, SysExitRecommendedError
 from rpdk.core.init import input_with_validation
-from rpdk.core.jsonutils.resolver import ContainerType, resolve_models
+from rpdk.core.jsonutils.resolver import (
+    UNDEFINED,
+    ContainerType,
+    ResolvedType,
+    resolve_models,
+)
 from rpdk.core.plugin_base import LanguagePlugin
 from rpdk.core.project import ARTIFACT_TYPE_HOOK
 from subprocess import PIPE, CalledProcessError, run as subprocess_run  # nosec
@@ -210,8 +215,48 @@ def generate(self, project):
         contents = template.render(support_lib_pkg=SUPPORT_LIB_PKG, models=models)
         project.overwrite(path, contents)
 
+        if project.artifact_type == ARTIFACT_TYPE_HOOK:
+            self._generate_target_models(project)
+
         LOG.debug("Generate complete")
 
+    def _generate_target_models(self, project):
+        target_model_dir = self.package_root / self.package_name / "target_models"
+
+        LOG.debug("Removing generated models: %s", target_model_dir)
+        shutil.rmtree(target_model_dir, ignore_errors=True)
+
+        target_model_dir.mkdir(parents=True, exist_ok=True)
+        template = self.env.get_template("target_model.py")
+
+        for target_type_name, target_info in project.target_info.items():
+            target_schema = target_info["Schema"]
+            target_namespace = [
+                s.lower() for s in target_type_name.split("::")
+            ]  # AWS::SQS::Queue -> awssqsqueue
+            target_name = "".join(
+                [s.capitalize() for s in target_namespace]
+            )  # awssqsqueue -> AwsSqsQueue
+            target_model_file = "{}.py".format(
+                "_".join(target_namespace)
+            )  # awssqsqueue -> aws_sqs_queue.py
+
+            models = resolve_models(target_schema, target_name)
+
+            # TODO: Remove once tagging is fully supported
+            if models.get(target_name, {}).get("Tags"):  # pragma: no cover
+                models[target_name]["Tags"] = ResolvedType(
+                    ContainerType.PRIMITIVE, UNDEFINED
+                )
+
+            path = target_model_dir / target_model_file
+            LOG.debug("Writing file: %s", path)
+
+            contents = template.render(
+                support_lib_pkg=SUPPORT_LIB_PKG, models=models, target_name=target_name
+            )
+            project.overwrite(path, contents)
+
     # pylint: disable=unused-argument
     # the argument "project" is not used here but is used in codegen.py of other plugins
     # this method is called in cloudformation-cli/src/rpdk/core/project.py

python/rpdk/python/templates/target_model.py

@@ -43,12 +43,8 @@ def _deserialize(
         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)
+        recast_object(cls, json_data, dataclasses)
         {% endif %}
         return cls(
             {% for name, type in properties.items() %}

tests/plugin/codegen_test.py

@@ -415,6 +415,9 @@ def test_generate_hook(hook_project):
     print("Project files: ", get_files_in_project(hook_project))
     assert files == {
         f"{os.path.join('src', 'foo_bar_baz', 'models.py')}",
+        f"{os.path.join('src', 'foo_bar_baz', 'target_models')}",
+        f"{os.path.join('src', 'foo_bar_baz', 'target_models', 'my_example_resource.py')}",  # noqa: B950 pylint: disable=line-too-long
+        f"{os.path.join('src', 'foo_bar_baz', 'target_models', 'my_other_resource.py')}",  # noqa: B950 pylint: disable=line-too-long
         "foo-bar-baz-configuration.json",
     }