in samcli/hook_packages/terraform/hooks/prepare/resource_linking.py [0:0]
def _link_using_linking_fields(self, cfn_resource: Dict) -> None:
"""
Depends on that all the child resources of the source resource are applied, and so we do not need to traverse
the terraform configuration to define the destination resource. We will depend on the actual values of the
linking fields, and find the destination resource that has the same value.
Parameters
----------
cfn_source_resource: Dict
A mapped CFN source resource
"""
# get the constant values of the linking field from the cfn_resource
values = cfn_resource.get("Properties", {}).get(self._resource_pair.cfn_link_field_name)
LOG.debug(
"Link the source resource %s using linking property %s that has the value %s",
cfn_resource,
self._resource_pair.cfn_link_field_name,
values,
)
# some resources can be linked to only one child resource like rest apis.
# make the resource values as a list to make processing easier.
if not isinstance(values, List):
values = [values]
# loop on the linking field values and call the Logical Id extractor function to extrac the destination resource
# logical id.
values = [
self._resource_pair.tf_destination_value_extractor_from_link_field_value_function(value) for value in values
]
# build map between the destination linking field property values, and resources' logical ids
expected_destinations_map = {
expected_destination.terraform_resource_type_prefix: expected_destination.terraform_attribute_name
for expected_destination in self._resource_pair.expected_destinations
}
child_resources_linking_attributes_logical_id_mapping = {}
for logical_id, destination_resource in self._resource_pair.destination_resource_tf.items():
destination_attribute = expected_destinations_map.get(f"{destination_resource.get('type', '')}.", "")
linking_attribute_value = destination_resource.get("values", {}).get(destination_attribute)
if linking_attribute_value:
child_resources_linking_attributes_logical_id_mapping[linking_attribute_value] = (
logical_id,
destination_resource.get("type", {}),
)
LOG.debug(
"The map between destination resources linking fields %s, and resources logical ids is %s",
", ".join(
[
expected_destination.terraform_attribute_name
for expected_destination in self._resource_pair.expected_destinations
]
),
child_resources_linking_attributes_logical_id_mapping,
)
dest_resources: List[ReferenceType] = [
(
LogicalIdReference(
value=child_resources_linking_attributes_logical_id_mapping[value][0],
resource_type=child_resources_linking_attributes_logical_id_mapping[value][1],
)
if value in child_resources_linking_attributes_logical_id_mapping
else ExistingResourceReference(value)
)
for value in values
]
if not dest_resources:
LOG.debug("Skipping linking call back, no destination resources discovered.")
return
LOG.debug("The value of the source resource linking field after mapping %s", dest_resources)
self._resource_pair.cfn_resource_update_call_back_function(cfn_resource, dest_resources)