feat: Attach connectors to source

samtranslator/translator/translator.py

@@ -124,6 +124,9 @@ def translate(
 
         self.sam_parser.parse(sam_template=sam_template, parameter_values=parameter_values, sam_plugins=sam_plugins)
 
+        updated_resources = add_embedded_connectors(sam_template=sam_template)
+        sam_template["Resources"].update(updated_resources)
+
         template = copy.deepcopy(sam_template)
         macro_resolver = ResourceTypeResolver(sam_resources)
         intrinsics_resolver = IntrinsicsResolver(parameter_values)
@@ -269,6 +272,64 @@ def _get_resources_to_iterate(
         return functions + statemachines + apis + others + connectors
 
 
+def make_error(msg: str) -> Exception:
+    return InvalidDocumentException([InvalidTemplateException(msg)])
+
+
+def add_embedded_connectors(sam_template: Dict[str, Any]) -> Any:
+    resources = sam_template.get("Resources", {})
+    connectors_list = []
+
+    # Loop through the resources in the template and see if any connectors have been attached
+    for source_logicalId, resource in resources.items():
+        if "Connectors" in resource:
+            if not isinstance(resource["Connectors"], dict):
+                raise make_error("'Connectors' must be a dict")
+
+            # Go through each of the connectors that have been attached and create a Serverless Connector resource
+            for connector_logicalId, connector_dict in resource["Connectors"].items():
+                if not isinstance(connector_dict, dict):
+                    raise make_error("Invalid 'Connectors' type. Must be a dict.")
+
+                data = get_generated_connector(source_logicalId, connector_logicalId, connector_dict)
+                # add to the list of generated connectors
+                connectors_list.append(data)
+
+            # delete connectors key, so it doesn't show up in the transformed template
+            del resource["Connectors"]
+
+    # go through the list of generated connectors and add it to resources dict
+    for connector_logicalId, connector_dict in connectors_list:
+        if connector_logicalId in resources:
+            raise make_error(
+                f"A resource with the id {connector_logicalId} already exists in the "
+                f"resource. Please use a different id for that resource.",
+            )
+        resources[connector_logicalId] = connector_dict
+
+    return resources
+
+
+def get_generated_connector(
+    source_logicalId: str, connector_logicalId: str, connector_dict: Dict[str, Any]
+) -> Tuple[str, Dict[str, Any]]:
+    connector = connector_dict
+    connector["Type"] = "AWS::Serverless::Connector"
+
+    # No need to raise an error for this instance as the error will be caught by the parser
+    if isinstance(connector["Properties"], dict):
+        connector["Properties"]["Source"] = {"Id": source_logicalId}
+
+        if "SourceReference" in connector["Properties"]:
+            if not isinstance(connector["Properties"]["SourceReference"], dict):
+                raise make_error(f"The SourceReference property for {connector_logicalId} must be a dict")
+
+            connector["Properties"]["Source"].update(connector["Properties"]["SourceReference"])
+            del connector["Properties"]["SourceReference"]
+
+    return connector_logicalId, connector
+
+
 def prepare_plugins(plugins: List[Any], parameters: Optional[Dict[str, Any]] = None) -> SamPlugins:
     """
     Creates & returns a plugins object with the given list of plugins installed. In addition to the given plugins,