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 TYPE_CHECKING
9import warnings
10
11from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error
12from azure.core.paging import ItemPaged
13from azure.core.pipeline import PipelineResponse
14from azure.core.pipeline.transport import HttpRequest, HttpResponse
15from azure.core.polling import LROPoller, NoPolling, PollingMethod
16from azure.mgmt.core.exceptions import ARMErrorFormat
17from azure.mgmt.core.polling.arm_polling import ARMPolling
18
19from .. import models as _models
20
21if TYPE_CHECKING:
22    # pylint: disable=unused-import,ungrouped-imports
23    from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar, Union
24
25    T = TypeVar('T')
26    ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
27
28class ServerTrustGroupsOperations(object):
29    """ServerTrustGroupsOperations operations.
30
31    You should not instantiate this class directly. Instead, you should create a Client instance that
32    instantiates it for you and attaches it as an attribute.
33
34    :ivar models: Alias to model classes used in this operation group.
35    :type models: ~azure.mgmt.sql.models
36    :param client: Client for service requests.
37    :param config: Configuration of service client.
38    :param serializer: An object model serializer.
39    :param deserializer: An object model deserializer.
40    """
41
42    models = _models
43
44    def __init__(self, client, config, serializer, deserializer):
45        self._client = client
46        self._serialize = serializer
47        self._deserialize = deserializer
48        self._config = config
49
50    def get(
51        self,
52        resource_group_name,  # type: str
53        location_name,  # type: str
54        server_trust_group_name,  # type: str
55        **kwargs  # type: Any
56    ):
57        # type: (...) -> "_models.ServerTrustGroup"
58        """Gets a server trust group.
59
60        :param resource_group_name: The name of the resource group that contains the resource. You can
61         obtain this value from the Azure Resource Manager API or the portal.
62        :type resource_group_name: str
63        :param location_name: The name of the region where the resource is located.
64        :type location_name: str
65        :param server_trust_group_name: The name of the server trust group.
66        :type server_trust_group_name: str
67        :keyword callable cls: A custom type or function that will be passed the direct response
68        :return: ServerTrustGroup, or the result of cls(response)
69        :rtype: ~azure.mgmt.sql.models.ServerTrustGroup
70        :raises: ~azure.core.exceptions.HttpResponseError
71        """
72        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ServerTrustGroup"]
73        error_map = {
74            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
75        }
76        error_map.update(kwargs.pop('error_map', {}))
77        api_version = "2020-11-01-preview"
78        accept = "application/json"
79
80        # Construct URL
81        url = self.get.metadata['url']  # type: ignore
82        path_format_arguments = {
83            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
84            'locationName': self._serialize.url("location_name", location_name, 'str'),
85            'serverTrustGroupName': self._serialize.url("server_trust_group_name", server_trust_group_name, 'str'),
86            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
87        }
88        url = self._client.format_url(url, **path_format_arguments)
89
90        # Construct parameters
91        query_parameters = {}  # type: Dict[str, Any]
92        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
93
94        # Construct headers
95        header_parameters = {}  # type: Dict[str, Any]
96        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
97
98        request = self._client.get(url, query_parameters, header_parameters)
99        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
100        response = pipeline_response.http_response
101
102        if response.status_code not in [200]:
103            map_error(status_code=response.status_code, response=response, error_map=error_map)
104            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
105
106        deserialized = self._deserialize('ServerTrustGroup', pipeline_response)
107
108        if cls:
109            return cls(pipeline_response, deserialized, {})
110
111        return deserialized
112    get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups/{serverTrustGroupName}'}  # type: ignore
113
114    def _create_or_update_initial(
115        self,
116        resource_group_name,  # type: str
117        location_name,  # type: str
118        server_trust_group_name,  # type: str
119        parameters,  # type: "_models.ServerTrustGroup"
120        **kwargs  # type: Any
121    ):
122        # type: (...) -> Optional["_models.ServerTrustGroup"]
123        cls = kwargs.pop('cls', None)  # type: ClsType[Optional["_models.ServerTrustGroup"]]
124        error_map = {
125            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
126        }
127        error_map.update(kwargs.pop('error_map', {}))
128        api_version = "2020-11-01-preview"
129        content_type = kwargs.pop("content_type", "application/json")
130        accept = "application/json"
131
132        # Construct URL
133        url = self._create_or_update_initial.metadata['url']  # type: ignore
134        path_format_arguments = {
135            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
136            'locationName': self._serialize.url("location_name", location_name, 'str'),
137            'serverTrustGroupName': self._serialize.url("server_trust_group_name", server_trust_group_name, 'str'),
138            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
139        }
140        url = self._client.format_url(url, **path_format_arguments)
141
142        # Construct parameters
143        query_parameters = {}  # type: Dict[str, Any]
144        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
145
146        # Construct headers
147        header_parameters = {}  # type: Dict[str, Any]
148        header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str')
149        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
150
151        body_content_kwargs = {}  # type: Dict[str, Any]
152        body_content = self._serialize.body(parameters, 'ServerTrustGroup')
153        body_content_kwargs['content'] = body_content
154        request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs)
155        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
156        response = pipeline_response.http_response
157
158        if response.status_code not in [200, 201, 202]:
159            map_error(status_code=response.status_code, response=response, error_map=error_map)
160            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
161
162        deserialized = None
163        if response.status_code == 200:
164            deserialized = self._deserialize('ServerTrustGroup', pipeline_response)
165
166        if response.status_code == 201:
167            deserialized = self._deserialize('ServerTrustGroup', pipeline_response)
168
169        if cls:
170            return cls(pipeline_response, deserialized, {})
171
172        return deserialized
173    _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups/{serverTrustGroupName}'}  # type: ignore
174
175    def begin_create_or_update(
176        self,
177        resource_group_name,  # type: str
178        location_name,  # type: str
179        server_trust_group_name,  # type: str
180        parameters,  # type: "_models.ServerTrustGroup"
181        **kwargs  # type: Any
182    ):
183        # type: (...) -> LROPoller["_models.ServerTrustGroup"]
184        """Creates or updates a server trust group.
185
186        :param resource_group_name: The name of the resource group that contains the resource. You can
187         obtain this value from the Azure Resource Manager API or the portal.
188        :type resource_group_name: str
189        :param location_name: The name of the region where the resource is located.
190        :type location_name: str
191        :param server_trust_group_name: The name of the server trust group.
192        :type server_trust_group_name: str
193        :param parameters: The server trust group parameters.
194        :type parameters: ~azure.mgmt.sql.models.ServerTrustGroup
195        :keyword callable cls: A custom type or function that will be passed the direct response
196        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
197        :keyword polling: By default, your polling method will be ARMPolling.
198         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
199        :paramtype polling: bool or ~azure.core.polling.PollingMethod
200        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
201        :return: An instance of LROPoller that returns either ServerTrustGroup or the result of cls(response)
202        :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.sql.models.ServerTrustGroup]
203        :raises ~azure.core.exceptions.HttpResponseError:
204        """
205        polling = kwargs.pop('polling', True)  # type: Union[bool, PollingMethod]
206        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ServerTrustGroup"]
207        lro_delay = kwargs.pop(
208            'polling_interval',
209            self._config.polling_interval
210        )
211        cont_token = kwargs.pop('continuation_token', None)  # type: Optional[str]
212        if cont_token is None:
213            raw_result = self._create_or_update_initial(
214                resource_group_name=resource_group_name,
215                location_name=location_name,
216                server_trust_group_name=server_trust_group_name,
217                parameters=parameters,
218                cls=lambda x,y,z: x,
219                **kwargs
220            )
221
222        kwargs.pop('error_map', None)
223        kwargs.pop('content_type', None)
224
225        def get_long_running_output(pipeline_response):
226            deserialized = self._deserialize('ServerTrustGroup', pipeline_response)
227
228            if cls:
229                return cls(pipeline_response, deserialized, {})
230            return deserialized
231
232        path_format_arguments = {
233            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
234            'locationName': self._serialize.url("location_name", location_name, 'str'),
235            'serverTrustGroupName': self._serialize.url("server_trust_group_name", server_trust_group_name, 'str'),
236            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
237        }
238
239        if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments,  **kwargs)
240        elif polling is False: polling_method = NoPolling()
241        else: polling_method = polling
242        if cont_token:
243            return LROPoller.from_continuation_token(
244                polling_method=polling_method,
245                continuation_token=cont_token,
246                client=self._client,
247                deserialization_callback=get_long_running_output
248            )
249        else:
250            return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
251    begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups/{serverTrustGroupName}'}  # type: ignore
252
253    def _delete_initial(
254        self,
255        resource_group_name,  # type: str
256        location_name,  # type: str
257        server_trust_group_name,  # type: str
258        **kwargs  # type: Any
259    ):
260        # type: (...) -> None
261        cls = kwargs.pop('cls', None)  # type: ClsType[None]
262        error_map = {
263            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
264        }
265        error_map.update(kwargs.pop('error_map', {}))
266        api_version = "2020-11-01-preview"
267
268        # Construct URL
269        url = self._delete_initial.metadata['url']  # type: ignore
270        path_format_arguments = {
271            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
272            'locationName': self._serialize.url("location_name", location_name, 'str'),
273            'serverTrustGroupName': self._serialize.url("server_trust_group_name", server_trust_group_name, 'str'),
274            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
275        }
276        url = self._client.format_url(url, **path_format_arguments)
277
278        # Construct parameters
279        query_parameters = {}  # type: Dict[str, Any]
280        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
281
282        # Construct headers
283        header_parameters = {}  # type: Dict[str, Any]
284
285        request = self._client.delete(url, query_parameters, header_parameters)
286        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
287        response = pipeline_response.http_response
288
289        if response.status_code not in [200, 202, 204]:
290            map_error(status_code=response.status_code, response=response, error_map=error_map)
291            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
292
293        if cls:
294            return cls(pipeline_response, None, {})
295
296    _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups/{serverTrustGroupName}'}  # type: ignore
297
298    def begin_delete(
299        self,
300        resource_group_name,  # type: str
301        location_name,  # type: str
302        server_trust_group_name,  # type: str
303        **kwargs  # type: Any
304    ):
305        # type: (...) -> LROPoller[None]
306        """Deletes a server trust group.
307
308        :param resource_group_name: The name of the resource group that contains the resource. You can
309         obtain this value from the Azure Resource Manager API or the portal.
310        :type resource_group_name: str
311        :param location_name: The name of the region where the resource is located.
312        :type location_name: str
313        :param server_trust_group_name: The name of the server trust group.
314        :type server_trust_group_name: str
315        :keyword callable cls: A custom type or function that will be passed the direct response
316        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
317        :keyword polling: By default, your polling method will be ARMPolling.
318         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
319        :paramtype polling: bool or ~azure.core.polling.PollingMethod
320        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
321        :return: An instance of LROPoller that returns either None or the result of cls(response)
322        :rtype: ~azure.core.polling.LROPoller[None]
323        :raises ~azure.core.exceptions.HttpResponseError:
324        """
325        polling = kwargs.pop('polling', True)  # type: Union[bool, PollingMethod]
326        cls = kwargs.pop('cls', None)  # type: ClsType[None]
327        lro_delay = kwargs.pop(
328            'polling_interval',
329            self._config.polling_interval
330        )
331        cont_token = kwargs.pop('continuation_token', None)  # type: Optional[str]
332        if cont_token is None:
333            raw_result = self._delete_initial(
334                resource_group_name=resource_group_name,
335                location_name=location_name,
336                server_trust_group_name=server_trust_group_name,
337                cls=lambda x,y,z: x,
338                **kwargs
339            )
340
341        kwargs.pop('error_map', None)
342        kwargs.pop('content_type', None)
343
344        def get_long_running_output(pipeline_response):
345            if cls:
346                return cls(pipeline_response, None, {})
347
348        path_format_arguments = {
349            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
350            'locationName': self._serialize.url("location_name", location_name, 'str'),
351            'serverTrustGroupName': self._serialize.url("server_trust_group_name", server_trust_group_name, 'str'),
352            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
353        }
354
355        if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments,  **kwargs)
356        elif polling is False: polling_method = NoPolling()
357        else: polling_method = polling
358        if cont_token:
359            return LROPoller.from_continuation_token(
360                polling_method=polling_method,
361                continuation_token=cont_token,
362                client=self._client,
363                deserialization_callback=get_long_running_output
364            )
365        else:
366            return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
367    begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups/{serverTrustGroupName}'}  # type: ignore
368
369    def list_by_location(
370        self,
371        resource_group_name,  # type: str
372        location_name,  # type: str
373        **kwargs  # type: Any
374    ):
375        # type: (...) -> Iterable["_models.ServerTrustGroupListResult"]
376        """Lists a server trust group.
377
378        :param resource_group_name: The name of the resource group that contains the resource. You can
379         obtain this value from the Azure Resource Manager API or the portal.
380        :type resource_group_name: str
381        :param location_name: The name of the region where the resource is located.
382        :type location_name: str
383        :keyword callable cls: A custom type or function that will be passed the direct response
384        :return: An iterator like instance of either ServerTrustGroupListResult or the result of cls(response)
385        :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ServerTrustGroupListResult]
386        :raises: ~azure.core.exceptions.HttpResponseError
387        """
388        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ServerTrustGroupListResult"]
389        error_map = {
390            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
391        }
392        error_map.update(kwargs.pop('error_map', {}))
393        api_version = "2020-11-01-preview"
394        accept = "application/json"
395
396        def prepare_request(next_link=None):
397            # Construct headers
398            header_parameters = {}  # type: Dict[str, Any]
399            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
400
401            if not next_link:
402                # Construct URL
403                url = self.list_by_location.metadata['url']  # type: ignore
404                path_format_arguments = {
405                    'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
406                    'locationName': self._serialize.url("location_name", location_name, 'str'),
407                    'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
408                }
409                url = self._client.format_url(url, **path_format_arguments)
410                # Construct parameters
411                query_parameters = {}  # type: Dict[str, Any]
412                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
413
414                request = self._client.get(url, query_parameters, header_parameters)
415            else:
416                url = next_link
417                query_parameters = {}  # type: Dict[str, Any]
418                request = self._client.get(url, query_parameters, header_parameters)
419            return request
420
421        def extract_data(pipeline_response):
422            deserialized = self._deserialize('ServerTrustGroupListResult', pipeline_response)
423            list_of_elem = deserialized.value
424            if cls:
425                list_of_elem = cls(list_of_elem)
426            return deserialized.next_link or None, iter(list_of_elem)
427
428        def get_next(next_link=None):
429            request = prepare_request(next_link)
430
431            pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
432            response = pipeline_response.http_response
433
434            if response.status_code not in [200]:
435                map_error(status_code=response.status_code, response=response, error_map=error_map)
436                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
437
438            return pipeline_response
439
440        return ItemPaged(
441            get_next, extract_data
442        )
443    list_by_location.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/serverTrustGroups'}  # type: ignore
444
445    def list_by_instance(
446        self,
447        resource_group_name,  # type: str
448        managed_instance_name,  # type: str
449        **kwargs  # type: Any
450    ):
451        # type: (...) -> Iterable["_models.ServerTrustGroupListResult"]
452        """Gets a server trust groups by instance name.
453
454        :param resource_group_name: The name of the resource group that contains the resource. You can
455         obtain this value from the Azure Resource Manager API or the portal.
456        :type resource_group_name: str
457        :param managed_instance_name: The name of the managed instance.
458        :type managed_instance_name: str
459        :keyword callable cls: A custom type or function that will be passed the direct response
460        :return: An iterator like instance of either ServerTrustGroupListResult or the result of cls(response)
461        :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ServerTrustGroupListResult]
462        :raises: ~azure.core.exceptions.HttpResponseError
463        """
464        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ServerTrustGroupListResult"]
465        error_map = {
466            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
467        }
468        error_map.update(kwargs.pop('error_map', {}))
469        api_version = "2020-11-01-preview"
470        accept = "application/json"
471
472        def prepare_request(next_link=None):
473            # Construct headers
474            header_parameters = {}  # type: Dict[str, Any]
475            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
476
477            if not next_link:
478                # Construct URL
479                url = self.list_by_instance.metadata['url']  # type: ignore
480                path_format_arguments = {
481                    'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
482                    'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'),
483                    'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
484                }
485                url = self._client.format_url(url, **path_format_arguments)
486                # Construct parameters
487                query_parameters = {}  # type: Dict[str, Any]
488                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
489
490                request = self._client.get(url, query_parameters, header_parameters)
491            else:
492                url = next_link
493                query_parameters = {}  # type: Dict[str, Any]
494                request = self._client.get(url, query_parameters, header_parameters)
495            return request
496
497        def extract_data(pipeline_response):
498            deserialized = self._deserialize('ServerTrustGroupListResult', pipeline_response)
499            list_of_elem = deserialized.value
500            if cls:
501                list_of_elem = cls(list_of_elem)
502            return deserialized.next_link or None, iter(list_of_elem)
503
504        def get_next(next_link=None):
505            request = prepare_request(next_link)
506
507            pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
508            response = pipeline_response.http_response
509
510            if response.status_code not in [200]:
511                map_error(status_code=response.status_code, response=response, error_map=error_map)
512                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
513
514            return pipeline_response
515
516        return ItemPaged(
517            get_next, extract_data
518        )
519    list_by_instance.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/managedInstances/{managedInstanceName}/serverTrustGroups'}  # type: ignore
520