def get_cloud_endpoints()

in samcli/lib/list/endpoints/endpoints_producer.py [0:0]
45 lines of code
16 McCabe index (conditional complexity)

    def get_cloud_endpoints(self, stacks: list) -> list:
        """
        Gets a list of cloud endpoints resources

        Parameters
        ----------
        stacks: list
            A list containing the local stack

        Returns
        -------
        endpoints_list: List[Any]
            A list of cloud endpoints resources
        """
        endpoints_list = []
        local_stack = stacks[0]
        local_stack_resources = local_stack.resources
        seen_endpoints = set()
        response = self.get_resources_info()
        response_domain_dict = get_response_domain_dict(response)
        custom_domain_substitute_dict = get_custom_domain_substitute_list(response, stacks, response_domain_dict)

        # Iterate over the deployed resources, collect relevant endpoint data for functions and APIGW resources
        for deployed_resource in response.get(STACK_RESOURCES, {}):
            if deployed_resource.get(RESOURCE_TYPE, "") in ENDPOINT_RESOURCE_TYPES:
                endpoint_function_url: Any
                paths_and_methods: Any
                endpoint_function_url = "-"
                paths_and_methods = "-"

                # Collect function URLs
                if deployed_resource.get(RESOURCE_TYPE, "") == AWS_LAMBDA_FUNCTION:
                    endpoint_function_url = self.get_function_url(deployed_resource.get(PHYSICAL_RESOURCE_ID, ""))

                # Collect APIGW endpoints and methods
                elif deployed_resource.get(RESOURCE_TYPE, "") in (AWS_APIGATEWAY_RESTAPI, AWS_APIGATEWAY_V2_API):
                    endpoint_function_url = self.get_api_gateway_endpoint(
                        deployed_resource, custom_domain_substitute_dict
                    )
                    paths_and_methods = get_methods_and_paths(
                        deployed_resource.get(LOGICAL_RESOURCE_ID, ""), local_stack
                    )

                endpoint_data = EndpointsDef(
                    LogicalResourceId=deployed_resource.get(LOGICAL_RESOURCE_ID, "-"),
                    PhysicalResourceId=deployed_resource.get(PHYSICAL_RESOURCE_ID, "-"),
                    CloudEndpoint=endpoint_function_url,
                    Methods=paths_and_methods,
                )
                endpoints_list.append(dataclasses.asdict(endpoint_data))
                seen_endpoints.add(deployed_resource.get(LOGICAL_RESOURCE_ID, ""))

        # Loop over resources all stack resources and collect data for resources not yet deployed
        for local_resource in local_stack_resources:
            local_resource_type = local_stack_resources.get(local_resource, {}).get("Type", "")
            paths_and_methods = "-"
            # Check if a resources has already been added to the endpoints list, if not, add it
            if local_resource_type in ENDPOINT_RESOURCE_TYPES and local_resource not in seen_endpoints:
                # We don't support function URLs locally, so this can only be APIGW endpoint data
                if local_resource_type in (AWS_APIGATEWAY_RESTAPI, AWS_APIGATEWAY_V2_API):
                    paths_and_methods = get_methods_and_paths(local_resource, local_stack)
                endpoint_data = EndpointsDef(
                    LogicalResourceId=local_resource,
                    PhysicalResourceId="-",
                    CloudEndpoint="-",
                    Methods=paths_and_methods,
                )
                endpoints_list.append(dataclasses.asdict(endpoint_data))

        return endpoints_list