# -*- coding: utf-8 -*- # @Time: 2023/8/29 11:06 class KYYAliECS: def __init__(self): # self.access_key_id = 'LTAI5tSX7wQnsx6dRfUc3Wfe' # self.access_key_secret = 'bWiuPh0eCWIaaXXpF9axsPVHX6kWzY' self.access_key_id = 'LTAI5tGYr9XoLaJD1vBR8Vu7' self.access_key_secret = 'jbFPoDHinlCOMXhIfQLag9nWX78IMb' self.sms_client = self.create_client() def create_client(self): """ 使用AK&SK初始化账号Client @param access_key_id: @param access_key_secret: @return: Client @throws Exception """ config = open_api_models.Config( access_key_id=self.access_key_id, access_key_secret=self.access_key_secret ) # Endpoint 请参考 https://api.aliyun.com/product/Ecs config.endpoint = f'ecs-cn-hangzhou.aliyuncs.com' return Ecs20140526Client(config) async def get_describe_regions(self, nss): """ instance_charge_type PrePaid：包年包月。此时，您必须确认自己的账号支持余额支付或者信用支付，否则将报错InvalidPayMethod PostPaid：按量付费 SpotWithPriceLimit: 枚举值 默认值为PostPaid。 resource_type: instance：ECS实例 disk：磁盘 reservedinstance：预留实例券 scu：存储容量单位包 默认值：instance accept_language: zh-CN：中文。 en-US：英文。 ja：日文。 默认值为zh-CN。 :return: """ describe_regions_request = ecs_20140526_models.DescribeRegionsRequest( instance_charge_type=nss.get('instance_charge_type'), resource_type=nss.get('resource_type'), accept_language=nss.get('accept_language') ) runtime = util_models.RuntimeOptions() resp = await self.sms_client.describe_regions_with_options_async(describe_regions_request, runtime) # ConsoleClient.log(UtilClient.to_jsonstring(resp)) return UtilClient.to_jsonstring(resp) async def describe_available_resource(self, ns={}): describe_available_resource_request = ecs_20140526_models.DescribeAvailableResourceRequest( region_id=ns.get('region_id'), memory=ns.get('memory'), resource_type=ns.get('resource_type'), scope=ns.get('scope'), cores=ns.get('cores'), network_category=ns.get('network_category'), data_disk_category=ns.get('data_disk_category'), system_disk_category=ns.get('system_disk_category'), instance_type=ns.get('instance_type'), dedicated_host_id=ns.get('dedicated_host_id'), io_optimized=ns.get('io_optimized'), zone_id=ns.get('zone_id'), destination_resource=ns.get('destination_resource'), spot_duration=ns.get('spot_duration'), spot_strategy=ns.get('spot_strategy'), instance_charge_type=ns.get('instance_charge_type') ) runtime = util_models.RuntimeOptions() try: # 复制代码运行请自行打印 API 的返回值 result = await self.sms_client.describe_available_resource_with_options_async(describe_available_resource_request, runtime) return result except Exception as error: # 如有需要，请打印 error return UtilClient.assert_as_string(error.message) async def describe_instance_types(self, ns={}): describe_instance_types_request = ecs_20140526_models.DescribeInstanceTypesRequest( instance_type_family=ns.get('instance_type_family'), instance_types=ns.get('instance_types'), minimum_cpu_core_count=ns.get('minimum_cpu_core_count'), maximum_cpu_core_count=ns.get('maximum_cpu_core_count'), minimum_memory_size=ns.get('minimum_memory_size'), maximum_memory_size=ns.get('maximum_memory_size'), minimum_gpuamount=ns.get('minimum_gpuamount'), maximum_gpuamount=ns.get('maximum_gpuamount'), gpuspec=ns.get('gpuspec'), instance_category=ns.get('instance_category'), cpu_architecture=ns.get('cpu_architecture'), minimum_cpu_speed_frequency=ns.get('minimum_cpu_speed_frequency'), maximum_cpu_speed_frequency=ns.get('maximum_cpu_speed_frequency'), minimum_cpu_turbo_frequency=ns.get('minimum_cpu_turbo_frequency'), maximum_cpu_turbo_frequency=ns.get('maximum_cpu_turbo_frequency'), physical_processor_model=ns.get('physical_processor_model'), instance_family_level=ns.get('instance_family_level'), minimum_instance_pps_rx=ns.get('minimum_instance_pps_rx'), minimum_instance_pps_tx=ns.get('minimum_instance_pps_tx'), minimum_instance_bandwidth_rx=ns.get('minimum_instance_bandwidth_rx'), minimum_instance_bandwidth_tx=ns.get('minimum_instance_bandwidth_tx'), minimum_primary_eni_queue_number=ns.get('minimum_primary_eni_queue_number'), minimum_secondary_eni_queue_number=ns.get('minimum_secondary_eni_queue_number'), minimum_eni_quantity=ns.get('minimum_eni_quantity'), minimum_queue_pair_number=ns.get('minimum_queue_pair_number'), minimum_eri_quantity=ns.get('minimum_eri_quantity'), minimum_eni_private_ip_address_quantity=ns.get('minimum_eni_private_ip_address_quantity'), minimum_eni_ipv_6address_quantity=ns.get('minimum_eni_ipv_6address_quantity'), minimum_local_storage_amount=ns.get('minimum_local_storage_amount'), minimum_local_storage_capacity=ns.get('minimum_local_storage_capacity'), minimum_disk_quantity=ns.get('minimum_disk_quantity'), local_storage_category=ns.get('local_storage_category'), nvme_support=ns.get('nvme_support'), minimum_baseline_credit=ns.get('minimum_baseline_credit'), minimum_initial_credit=ns.get('minimum_initial_credit'), max_results=ns.get('max_results'), next_token=ns.get('next_token'), ) runtime = util_models.RuntimeOptions() try: # 复制代码运行请自行打印 API 的返回值 result = await self.sms_client.describe_instance_types_with_options_async(describe_instance_types_request, runtime) return result except Exception as error: # 如有需要，请打印 error return UtilClient.assert_as_string(error.message) async def describe_price(self, ns={}): scheduler_options = ecs_20140526_models.DescribePriceRequestSchedulerOptions( dedicated_host_id='dh-bp67acfmxazb4p****' ) data_disk_0 = ecs_20140526_models.DescribePriceRequestDataDisk( category='cloud_ssd', size=3000, performance_level='PL1' ) # 快速购买 默认存在 system_disk = ecs_20140526_models.DescribePriceRequestSystemDisk( category=ns.get('system_disk').get('category'), size=ns.get('system_disk').get('size') ) describe_price_request = ecs_20140526_models.DescribePriceRequest( region_id=ns.get('region_id'), resource_type=ns.get('resource_type'), image_id=ns.get('image_id'), instance_type=ns.get('instance_type'), dedicated_host_type=ns.get('dedicated_host_type'), io_optimized=ns.get('io_optimized'), instance_network_type=ns.get('instance_network_type'), internet_charge_type=ns.get('internet_charge_type'), internet_max_bandwidth_out=ns.get('internet_max_bandwidth_out'), system_disk=system_disk, data_disk=ns.get('data_disk'), period=ns.get('period'), price_unit=ns.get('price_unit'), amount=ns.get('amount'), offering_type=ns.get('offering_type'), instance_amount=ns.get('instance_amount'), scope=ns.get('scope'), platform=ns.get('platform'), capacity=ns.get('capacity'), assurance_times=ns.get('assurance_times'), instance_cpu_core_count=ns.get('instance_cpu_core_count'), isp=ns.get('isp'), instance_type_list=ns.get('instance_type_list'), spot_strategy=ns.get('spot_strategy'), spot_duration=ns.get('spot_duration'), zone_id=ns.get('zone_id'), scheduler_options=ns.get('scheduler_options') ) runtime = util_models.RuntimeOptions() try: result = await self.sms_client.describe_price_with_options_async(describe_price_request, runtime) return result except Exception as error: return UtilClient.assert_as_string(error.message) async def describe_images(self, ns={}): describe_images_request = ecs_20140526_models.DescribeImagesRequest( region_id='cn-qingdao' # region_id=ns.get('region_id'), # show_expired=ns.get('show_expired'), # is_support_io_optimized=ns.get('is_support_io_optimized'), # is_support_cloudinit=ns.get('is_support_cloudinit'), # ostype=ns.get('ostype'), # architecture=ns.get('architecture'), # usage=ns.get('usage'), # dry_run=ns.get('dry_run'), # action_type=ns.get('action_type'), # resource_group_id=ns.get('resource_group_id'), # is_public=ns.get('is_public'), # image_owner_id=ns.get('image_owner_id'), # page_number=ns.get('page_number'), # page_size=ns.get('page_size'), ) runtime = util_models.RuntimeOptions() try: result = await self.sms_client.describe_images_with_options_async(describe_images_request, runtime) return result except Exception as error: # 如有需要，请打印 error return UtilClient.assert_as_string(error.message) async def DescribeRegions(ns={}): """ 获取可用区域 :param ns: :return: """ # ns = { # 'instance_charge_type': 'PrePaid', # 'resource_type': 'instance', # 'accept_language': 'zh-CN' # # 'resource_type': 'disk' # } nss = { 'instance_charge_type': ns.get('instance_charge_type'), 'resource_type': ns.get('resource_type'), 'accept_language': ns.get('accept_language') } regions = await KYYAliECS().get_describe_regions(nss) result = json.loads(regions)['body'] result['Regions']['Region_filter'] = {'Asia_China': [], 'Asia_Other': [], 'Europe_America': [], 'Middle_East': []} for content in result['Regions']['Region']: if 'cn' in content['RegionId']: result['Regions']['Region_filter']['Asia_China'].append(content) elif 'ap' in content['RegionId']: result['Regions']['Region_filter']['Asia_Other'].append(content) elif 'en' in content['RegionId'] or 'us' in content['RegionId']: result['Regions']['Region_filter']['Europe_America'].append(content) elif 'me' in content['RegionId']: result['Regions']['Region_filter']['Middle_East'].append(content) print(result) return result ret = await DescribeRegions(params_kw) return ret