def _get_embedded_connectors()

in samtranslator/translator/translator.py [0:0]

• 37 lines of code
• 10 McCabe index (conditional complexity)

    def _get_embedded_connectors(self, resources: Dict[str, Any]) -> List[Resource]:
        """
        Loops through the SAM Template resources to find any connectors that have been attached to the resources.
        Converts those attached connectors into Connector resources and returns a list of them

        :param dict resources: Dict of resources from the SAM template
        :return List[SamConnector]: List of the generated SAM Connectors
        """
        connectors = []

        # Loop through the resources in the template and see if any connectors have been attached
        for source_logical_id, resource in resources.items():
            if "Connectors" not in resource:
                continue
            try:
                sam_expect(
                    resource.get("Connectors"),
                    source_logical_id,
                    f"{source_logical_id}.Connectors",
                    is_resource_attribute=True,
                ).to_be_a_map()
            except InvalidResourceException as e:
                self.document_errors.append(e)
                continue
            for connector_logical_id, connector_dict in resource["Connectors"].items():
                try:
                    full_connector_logical_id = source_logical_id + connector_logical_id
                    # can't use sam_expect since this is neither a property nor a resource attribute
                    if not isinstance(connector_dict, dict):
                        raise InvalidResourceException(
                            full_connector_logical_id,
                            f"{source_logical_id}.{full_connector_logical_id} should be a map.",
                        )

                    generated_connector = self._get_generated_connector(
                        source_logical_id,
                        full_connector_logical_id,
                        connector_logical_id,
                        connector_dict,
                    )

                    if not verify_unique_logical_id(generated_connector, resources):
                        raise DuplicateLogicalIdException(
                            source_logical_id, full_connector_logical_id, generated_connector.resource_type
                        )
                    connectors.append(generated_connector)
                except (InvalidResourceException, DuplicateLogicalIdException) as e:
                    self.document_errors.append(e)

        return connectors