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