1# coding=utf-8 2# -------------------------------------------------------------------------- 3# Copyright (c) Microsoft Corporation. All rights reserved. 4# Licensed under the MIT License. See License.txt in the project root for license information. 5# Code generated by Microsoft (R) AutoRest Code Generator. 6# Changes may cause incorrect behavior and will be lost if the code is regenerated. 7# -------------------------------------------------------------------------- 8from typing import Any, AsyncIterable, Callable, Dict, Generic, Optional, TypeVar 9import warnings 10 11from azure.core.async_paging import AsyncItemPaged, AsyncList 12from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 15from azure.mgmt.core.exceptions import ARMErrorFormat 16 17from ... import models as _models 18 19T = TypeVar('T') 20ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 21 22class CloudServiceOperatingSystemsOperations: 23 """CloudServiceOperatingSystemsOperations async operations. 24 25 You should not instantiate this class directly. Instead, you should create a Client instance that 26 instantiates it for you and attaches it as an attribute. 27 28 :ivar models: Alias to model classes used in this operation group. 29 :type models: ~azure.mgmt.compute.v2021_03_01.models 30 :param client: Client for service requests. 31 :param config: Configuration of service client. 32 :param serializer: An object model serializer. 33 :param deserializer: An object model deserializer. 34 """ 35 36 models = _models 37 38 def __init__(self, client, config, serializer, deserializer) -> None: 39 self._client = client 40 self._serialize = serializer 41 self._deserialize = deserializer 42 self._config = config 43 44 async def get_os_version( 45 self, 46 location: str, 47 os_version_name: str, 48 **kwargs: Any 49 ) -> "_models.OSVersion": 50 """Gets properties of a guest operating system version that can be specified in the XML service 51 configuration (.cscfg) for a cloud service. 52 53 :param location: Name of the location that the OS version pertains to. 54 :type location: str 55 :param os_version_name: Name of the OS version. 56 :type os_version_name: str 57 :keyword callable cls: A custom type or function that will be passed the direct response 58 :return: OSVersion, or the result of cls(response) 59 :rtype: ~azure.mgmt.compute.v2021_03_01.models.OSVersion 60 :raises: ~azure.core.exceptions.HttpResponseError 61 """ 62 cls = kwargs.pop('cls', None) # type: ClsType["_models.OSVersion"] 63 error_map = { 64 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 65 } 66 error_map.update(kwargs.pop('error_map', {})) 67 api_version = "2021-03-01" 68 accept = "application/json" 69 70 # Construct URL 71 url = self.get_os_version.metadata['url'] # type: ignore 72 path_format_arguments = { 73 'location': self._serialize.url("location", location, 'str'), 74 'osVersionName': self._serialize.url("os_version_name", os_version_name, 'str'), 75 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 76 } 77 url = self._client.format_url(url, **path_format_arguments) 78 79 # Construct parameters 80 query_parameters = {} # type: Dict[str, Any] 81 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 82 83 # Construct headers 84 header_parameters = {} # type: Dict[str, Any] 85 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 86 87 request = self._client.get(url, query_parameters, header_parameters) 88 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 89 response = pipeline_response.http_response 90 91 if response.status_code not in [200]: 92 map_error(status_code=response.status_code, response=response, error_map=error_map) 93 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 94 95 deserialized = self._deserialize('OSVersion', pipeline_response) 96 97 if cls: 98 return cls(pipeline_response, deserialized, {}) 99 100 return deserialized 101 get_os_version.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Compute/locations/{location}/cloudServiceOsVersions/{osVersionName}'} # type: ignore 102 103 def list_os_versions( 104 self, 105 location: str, 106 **kwargs: Any 107 ) -> AsyncIterable["_models.OSVersionListResult"]: 108 """Gets a list of all guest operating system versions available to be specified in the XML service 109 configuration (.cscfg) for a cloud service. Use nextLink property in the response to get the 110 next page of OS versions. Do this till nextLink is null to fetch all the OS versions. 111 112 :param location: Name of the location that the OS versions pertain to. 113 :type location: str 114 :keyword callable cls: A custom type or function that will be passed the direct response 115 :return: An iterator like instance of either OSVersionListResult or the result of cls(response) 116 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.compute.v2021_03_01.models.OSVersionListResult] 117 :raises: ~azure.core.exceptions.HttpResponseError 118 """ 119 cls = kwargs.pop('cls', None) # type: ClsType["_models.OSVersionListResult"] 120 error_map = { 121 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 122 } 123 error_map.update(kwargs.pop('error_map', {})) 124 api_version = "2021-03-01" 125 accept = "application/json" 126 127 def prepare_request(next_link=None): 128 # Construct headers 129 header_parameters = {} # type: Dict[str, Any] 130 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 131 132 if not next_link: 133 # Construct URL 134 url = self.list_os_versions.metadata['url'] # type: ignore 135 path_format_arguments = { 136 'location': self._serialize.url("location", location, 'str'), 137 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 138 } 139 url = self._client.format_url(url, **path_format_arguments) 140 # Construct parameters 141 query_parameters = {} # type: Dict[str, Any] 142 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 143 144 request = self._client.get(url, query_parameters, header_parameters) 145 else: 146 url = next_link 147 query_parameters = {} # type: Dict[str, Any] 148 request = self._client.get(url, query_parameters, header_parameters) 149 return request 150 151 async def extract_data(pipeline_response): 152 deserialized = self._deserialize('OSVersionListResult', pipeline_response) 153 list_of_elem = deserialized.value 154 if cls: 155 list_of_elem = cls(list_of_elem) 156 return deserialized.next_link or None, AsyncList(list_of_elem) 157 158 async def get_next(next_link=None): 159 request = prepare_request(next_link) 160 161 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 162 response = pipeline_response.http_response 163 164 if response.status_code not in [200]: 165 map_error(status_code=response.status_code, response=response, error_map=error_map) 166 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 167 168 return pipeline_response 169 170 return AsyncItemPaged( 171 get_next, extract_data 172 ) 173 list_os_versions.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Compute/locations/{location}/cloudServiceOsVersions'} # type: ignore 174 175 async def get_os_family( 176 self, 177 location: str, 178 os_family_name: str, 179 **kwargs: Any 180 ) -> "_models.OSFamily": 181 """Gets properties of a guest operating system family that can be specified in the XML service 182 configuration (.cscfg) for a cloud service. 183 184 :param location: Name of the location that the OS family pertains to. 185 :type location: str 186 :param os_family_name: Name of the OS family. 187 :type os_family_name: str 188 :keyword callable cls: A custom type or function that will be passed the direct response 189 :return: OSFamily, or the result of cls(response) 190 :rtype: ~azure.mgmt.compute.v2021_03_01.models.OSFamily 191 :raises: ~azure.core.exceptions.HttpResponseError 192 """ 193 cls = kwargs.pop('cls', None) # type: ClsType["_models.OSFamily"] 194 error_map = { 195 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 196 } 197 error_map.update(kwargs.pop('error_map', {})) 198 api_version = "2021-03-01" 199 accept = "application/json" 200 201 # Construct URL 202 url = self.get_os_family.metadata['url'] # type: ignore 203 path_format_arguments = { 204 'location': self._serialize.url("location", location, 'str'), 205 'osFamilyName': self._serialize.url("os_family_name", os_family_name, 'str'), 206 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 207 } 208 url = self._client.format_url(url, **path_format_arguments) 209 210 # Construct parameters 211 query_parameters = {} # type: Dict[str, Any] 212 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 213 214 # Construct headers 215 header_parameters = {} # type: Dict[str, Any] 216 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 217 218 request = self._client.get(url, query_parameters, header_parameters) 219 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 220 response = pipeline_response.http_response 221 222 if response.status_code not in [200]: 223 map_error(status_code=response.status_code, response=response, error_map=error_map) 224 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 225 226 deserialized = self._deserialize('OSFamily', pipeline_response) 227 228 if cls: 229 return cls(pipeline_response, deserialized, {}) 230 231 return deserialized 232 get_os_family.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Compute/locations/{location}/cloudServiceOsFamilies/{osFamilyName}'} # type: ignore 233 234 def list_os_families( 235 self, 236 location: str, 237 **kwargs: Any 238 ) -> AsyncIterable["_models.OSFamilyListResult"]: 239 """Gets a list of all guest operating system families available to be specified in the XML service 240 configuration (.cscfg) for a cloud service. Use nextLink property in the response to get the 241 next page of OS Families. Do this till nextLink is null to fetch all the OS Families. 242 243 :param location: Name of the location that the OS families pertain to. 244 :type location: str 245 :keyword callable cls: A custom type or function that will be passed the direct response 246 :return: An iterator like instance of either OSFamilyListResult or the result of cls(response) 247 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.compute.v2021_03_01.models.OSFamilyListResult] 248 :raises: ~azure.core.exceptions.HttpResponseError 249 """ 250 cls = kwargs.pop('cls', None) # type: ClsType["_models.OSFamilyListResult"] 251 error_map = { 252 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 253 } 254 error_map.update(kwargs.pop('error_map', {})) 255 api_version = "2021-03-01" 256 accept = "application/json" 257 258 def prepare_request(next_link=None): 259 # Construct headers 260 header_parameters = {} # type: Dict[str, Any] 261 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 262 263 if not next_link: 264 # Construct URL 265 url = self.list_os_families.metadata['url'] # type: ignore 266 path_format_arguments = { 267 'location': self._serialize.url("location", location, 'str'), 268 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 269 } 270 url = self._client.format_url(url, **path_format_arguments) 271 # Construct parameters 272 query_parameters = {} # type: Dict[str, Any] 273 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 274 275 request = self._client.get(url, query_parameters, header_parameters) 276 else: 277 url = next_link 278 query_parameters = {} # type: Dict[str, Any] 279 request = self._client.get(url, query_parameters, header_parameters) 280 return request 281 282 async def extract_data(pipeline_response): 283 deserialized = self._deserialize('OSFamilyListResult', pipeline_response) 284 list_of_elem = deserialized.value 285 if cls: 286 list_of_elem = cls(list_of_elem) 287 return deserialized.next_link or None, AsyncList(list_of_elem) 288 289 async def get_next(next_link=None): 290 request = prepare_request(next_link) 291 292 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 293 response = pipeline_response.http_response 294 295 if response.status_code not in [200]: 296 map_error(status_code=response.status_code, response=response, error_map=error_map) 297 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 298 299 return pipeline_response 300 301 return AsyncItemPaged( 302 get_next, extract_data 303 ) 304 list_os_families.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Compute/locations/{location}/cloudServiceOsFamilies'} # type: ignore 305