def _call_location_service()

in aliyun-python-sdk-core/aliyunsdkcore/endpoint/location_service_endpoint_resolver.py [0:0]

• 35 lines of code
• 12 McCabe index (conditional complexity)

    def _call_location_service(self, key, raw_request):
        request = DescribeEndpointsRequest()
        request.set_protocol_type("https")
        request.set_accept_format("json")
        request.set_Id(raw_request.region_id)
        request.set_ServiceCode(raw_request.location_service_code)
        request.set_Type(raw_request.endpoint_type)
        request.endpoint = self._location_service_endpoint

        try:
            response = self._client.do_action_with_exception(request)
        except ServerException as e:
            if "InvalidRegionId" == e.get_error_code() and \
               "The specified region does not exist." == e.get_error_msg():
                # No such region`
                self._invalid_region_ids.add(raw_request.region_id)
                self.put_endpoint_entry(key, None)
                return
            elif "Illegal Parameter" == e.get_error_code() and \
                 "Please check the parameters" == e.get_error_msg():
                # No such product
                self._invalid_product_codes.add(raw_request.product_code_lower)
                self.put_endpoint_entry(key, None)
                return
            else:
                raise e

        # As long as code gets here
        # the product code and the region id is valid
        # the endpoint can be still not found
        self._valid_product_codes.add(raw_request.product_code_lower)
        self._valid_region_ids.add(raw_request.region_id)

        found_flag = False
        body = json.loads(response.decode('utf-8'))
        for item in body["Endpoints"]["Endpoint"]:

            # Location return data has a typo: SerivceCode
            # We still try to expect ServiceCode in case this typo would be fixed in the future
            service_code = item.get("ServiceCode") or item.get("SerivceCode")

            if service_code and item.get("Type") == raw_request.endpoint_type:
                found_flag = True
                self.put_endpoint_entry(key, item.get("Endpoint"))
                break

        if not found_flag:
            self.put_endpoint_entry(key, None)