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