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 ResourcesOperations(object): 29 """ResourcesOperations 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.resource.resources.v2019_05_01.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 list_by_resource_group( 51 self, 52 resource_group_name, # type: str 53 filter=None, # type: Optional[str] 54 expand=None, # type: Optional[str] 55 top=None, # type: Optional[int] 56 **kwargs # type: Any 57 ): 58 # type: (...) -> Iterable["_models.ResourceListResult"] 59 """Get all the resources for a resource group. 60 61 :param resource_group_name: The resource group with the resources to get. 62 :type resource_group_name: str 63 :param filter: The filter to apply on the operation.:code:`<br>`:code:`<br>`The properties you 64 can use for eq (equals) or ne (not equals) are: location, resourceType, name, resourceGroup, 65 identity, identity/principalId, plan, plan/publisher, plan/product, plan/name, plan/version, 66 and plan/promotionCode.:code:`<br>`:code:`<br>`For example, to filter by a resource type, use: 67 $filter=resourceType eq 'Microsoft.Network/virtualNetworks':code:`<br>`:code:`<br>`You can use 68 substringof(value, property) in the filter. The properties you can use for substring are: name 69 and resourceGroup.:code:`<br>`:code:`<br>`For example, to get all resources with 'demo' 70 anywhere in the name, use: $filter=substringof('demo', name):code:`<br>`:code:`<br>`You can 71 link more than one substringof together by adding and/or operators.:code:`<br>`:code:`<br>`You 72 can filter by tag names and values. For example, to filter for a tag name and value, use 73 $filter=tagName eq 'tag1' and tagValue eq 'Value1':code:`<br>`:code:`<br>`You can use some 74 properties together when filtering. The combinations you can use are: substringof and/or 75 resourceType, plan and plan/publisher and plan/name, identity and identity/principalId. 76 :type filter: str 77 :param expand: Comma-separated list of additional properties to be included in the response. 78 Valid values include ``createdTime``\ , ``changedTime`` and ``provisioningState``. For example, 79 ``$expand=createdTime,changedTime``. 80 :type expand: str 81 :param top: The number of results to return. If null is passed, returns all resources. 82 :type top: int 83 :keyword callable cls: A custom type or function that will be passed the direct response 84 :return: An iterator like instance of either ResourceListResult or the result of cls(response) 85 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2019_05_01.models.ResourceListResult] 86 :raises: ~azure.core.exceptions.HttpResponseError 87 """ 88 cls = kwargs.pop('cls', None) # type: ClsType["_models.ResourceListResult"] 89 error_map = { 90 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 91 } 92 error_map.update(kwargs.pop('error_map', {})) 93 api_version = "2019-05-01" 94 accept = "application/json" 95 96 def prepare_request(next_link=None): 97 # Construct headers 98 header_parameters = {} # type: Dict[str, Any] 99 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 100 101 if not next_link: 102 # Construct URL 103 url = self.list_by_resource_group.metadata['url'] # type: ignore 104 path_format_arguments = { 105 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 106 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 107 } 108 url = self._client.format_url(url, **path_format_arguments) 109 # Construct parameters 110 query_parameters = {} # type: Dict[str, Any] 111 if filter is not None: 112 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str') 113 if expand is not None: 114 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 115 if top is not None: 116 query_parameters['$top'] = self._serialize.query("top", top, 'int') 117 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 118 119 request = self._client.get(url, query_parameters, header_parameters) 120 else: 121 url = next_link 122 query_parameters = {} # type: Dict[str, Any] 123 request = self._client.get(url, query_parameters, header_parameters) 124 return request 125 126 def extract_data(pipeline_response): 127 deserialized = self._deserialize('ResourceListResult', pipeline_response) 128 list_of_elem = deserialized.value 129 if cls: 130 list_of_elem = cls(list_of_elem) 131 return deserialized.next_link or None, iter(list_of_elem) 132 133 def get_next(next_link=None): 134 request = prepare_request(next_link) 135 136 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 137 response = pipeline_response.http_response 138 139 if response.status_code not in [200]: 140 map_error(status_code=response.status_code, response=response, error_map=error_map) 141 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 142 143 return pipeline_response 144 145 return ItemPaged( 146 get_next, extract_data 147 ) 148 list_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/resources'} # type: ignore 149 150 def _move_resources_initial( 151 self, 152 source_resource_group_name, # type: str 153 parameters, # type: "_models.ResourcesMoveInfo" 154 **kwargs # type: Any 155 ): 156 # type: (...) -> None 157 cls = kwargs.pop('cls', None) # type: ClsType[None] 158 error_map = { 159 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 160 } 161 error_map.update(kwargs.pop('error_map', {})) 162 api_version = "2019-05-01" 163 content_type = kwargs.pop("content_type", "application/json") 164 165 # Construct URL 166 url = self._move_resources_initial.metadata['url'] # type: ignore 167 path_format_arguments = { 168 'sourceResourceGroupName': self._serialize.url("source_resource_group_name", source_resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 169 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 170 } 171 url = self._client.format_url(url, **path_format_arguments) 172 173 # Construct parameters 174 query_parameters = {} # type: Dict[str, Any] 175 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 176 177 # Construct headers 178 header_parameters = {} # type: Dict[str, Any] 179 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 180 181 body_content_kwargs = {} # type: Dict[str, Any] 182 body_content = self._serialize.body(parameters, 'ResourcesMoveInfo') 183 body_content_kwargs['content'] = body_content 184 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 185 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 186 response = pipeline_response.http_response 187 188 if response.status_code not in [202, 204]: 189 map_error(status_code=response.status_code, response=response, error_map=error_map) 190 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 191 192 if cls: 193 return cls(pipeline_response, None, {}) 194 195 _move_resources_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{sourceResourceGroupName}/moveResources'} # type: ignore 196 197 def begin_move_resources( 198 self, 199 source_resource_group_name, # type: str 200 parameters, # type: "_models.ResourcesMoveInfo" 201 **kwargs # type: Any 202 ): 203 # type: (...) -> LROPoller[None] 204 """Moves resources from one resource group to another resource group. 205 206 The resources to move must be in the same source resource group. The target resource group may 207 be in a different subscription. When moving resources, both the source group and the target 208 group are locked for the duration of the operation. Write and delete operations are blocked on 209 the groups until the move completes. 210 211 :param source_resource_group_name: The name of the resource group containing the resources to 212 move. 213 :type source_resource_group_name: str 214 :param parameters: Parameters for moving resources. 215 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.ResourcesMoveInfo 216 :keyword callable cls: A custom type or function that will be passed the direct response 217 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 218 :keyword polling: By default, your polling method will be ARMPolling. 219 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 220 :paramtype polling: bool or ~azure.core.polling.PollingMethod 221 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 222 :return: An instance of LROPoller that returns either None or the result of cls(response) 223 :rtype: ~azure.core.polling.LROPoller[None] 224 :raises ~azure.core.exceptions.HttpResponseError: 225 """ 226 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 227 cls = kwargs.pop('cls', None) # type: ClsType[None] 228 lro_delay = kwargs.pop( 229 'polling_interval', 230 self._config.polling_interval 231 ) 232 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 233 if cont_token is None: 234 raw_result = self._move_resources_initial( 235 source_resource_group_name=source_resource_group_name, 236 parameters=parameters, 237 cls=lambda x,y,z: x, 238 **kwargs 239 ) 240 241 kwargs.pop('error_map', None) 242 kwargs.pop('content_type', None) 243 244 def get_long_running_output(pipeline_response): 245 if cls: 246 return cls(pipeline_response, None, {}) 247 248 path_format_arguments = { 249 'sourceResourceGroupName': self._serialize.url("source_resource_group_name", source_resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 250 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 251 } 252 253 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 254 elif polling is False: polling_method = NoPolling() 255 else: polling_method = polling 256 if cont_token: 257 return LROPoller.from_continuation_token( 258 polling_method=polling_method, 259 continuation_token=cont_token, 260 client=self._client, 261 deserialization_callback=get_long_running_output 262 ) 263 else: 264 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 265 begin_move_resources.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{sourceResourceGroupName}/moveResources'} # type: ignore 266 267 def _validate_move_resources_initial( 268 self, 269 source_resource_group_name, # type: str 270 parameters, # type: "_models.ResourcesMoveInfo" 271 **kwargs # type: Any 272 ): 273 # type: (...) -> None 274 cls = kwargs.pop('cls', None) # type: ClsType[None] 275 error_map = { 276 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 277 } 278 error_map.update(kwargs.pop('error_map', {})) 279 api_version = "2019-05-01" 280 content_type = kwargs.pop("content_type", "application/json") 281 282 # Construct URL 283 url = self._validate_move_resources_initial.metadata['url'] # type: ignore 284 path_format_arguments = { 285 'sourceResourceGroupName': self._serialize.url("source_resource_group_name", source_resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 286 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 287 } 288 url = self._client.format_url(url, **path_format_arguments) 289 290 # Construct parameters 291 query_parameters = {} # type: Dict[str, Any] 292 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 293 294 # Construct headers 295 header_parameters = {} # type: Dict[str, Any] 296 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 297 298 body_content_kwargs = {} # type: Dict[str, Any] 299 body_content = self._serialize.body(parameters, 'ResourcesMoveInfo') 300 body_content_kwargs['content'] = body_content 301 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 302 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 303 response = pipeline_response.http_response 304 305 if response.status_code not in [202, 204, 409]: 306 map_error(status_code=response.status_code, response=response, error_map=error_map) 307 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 308 309 if cls: 310 return cls(pipeline_response, None, {}) 311 312 _validate_move_resources_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{sourceResourceGroupName}/validateMoveResources'} # type: ignore 313 314 def begin_validate_move_resources( 315 self, 316 source_resource_group_name, # type: str 317 parameters, # type: "_models.ResourcesMoveInfo" 318 **kwargs # type: Any 319 ): 320 # type: (...) -> LROPoller[None] 321 """Validates whether resources can be moved from one resource group to another resource group. 322 323 This operation checks whether the specified resources can be moved to the target. The resources 324 to move must be in the same source resource group. The target resource group may be in a 325 different subscription. If validation succeeds, it returns HTTP response code 204 (no content). 326 If validation fails, it returns HTTP response code 409 (Conflict) with an error message. 327 Retrieve the URL in the Location header value to check the result of the long-running 328 operation. 329 330 :param source_resource_group_name: The name of the resource group containing the resources to 331 validate for move. 332 :type source_resource_group_name: str 333 :param parameters: Parameters for moving resources. 334 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.ResourcesMoveInfo 335 :keyword callable cls: A custom type or function that will be passed the direct response 336 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 337 :keyword polling: By default, your polling method will be ARMPolling. 338 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 339 :paramtype polling: bool or ~azure.core.polling.PollingMethod 340 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 341 :return: An instance of LROPoller that returns either None or the result of cls(response) 342 :rtype: ~azure.core.polling.LROPoller[None] 343 :raises ~azure.core.exceptions.HttpResponseError: 344 """ 345 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 346 cls = kwargs.pop('cls', None) # type: ClsType[None] 347 lro_delay = kwargs.pop( 348 'polling_interval', 349 self._config.polling_interval 350 ) 351 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 352 if cont_token is None: 353 raw_result = self._validate_move_resources_initial( 354 source_resource_group_name=source_resource_group_name, 355 parameters=parameters, 356 cls=lambda x,y,z: x, 357 **kwargs 358 ) 359 360 kwargs.pop('error_map', None) 361 kwargs.pop('content_type', None) 362 363 def get_long_running_output(pipeline_response): 364 if cls: 365 return cls(pipeline_response, None, {}) 366 367 path_format_arguments = { 368 'sourceResourceGroupName': self._serialize.url("source_resource_group_name", source_resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 369 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 370 } 371 372 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 373 elif polling is False: polling_method = NoPolling() 374 else: polling_method = polling 375 if cont_token: 376 return LROPoller.from_continuation_token( 377 polling_method=polling_method, 378 continuation_token=cont_token, 379 client=self._client, 380 deserialization_callback=get_long_running_output 381 ) 382 else: 383 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 384 begin_validate_move_resources.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{sourceResourceGroupName}/validateMoveResources'} # type: ignore 385 386 def list( 387 self, 388 filter=None, # type: Optional[str] 389 expand=None, # type: Optional[str] 390 top=None, # type: Optional[int] 391 **kwargs # type: Any 392 ): 393 # type: (...) -> Iterable["_models.ResourceListResult"] 394 """Get all the resources in a subscription. 395 396 :param filter: The filter to apply on the operation.:code:`<br>`:code:`<br>`The properties you 397 can use for eq (equals) or ne (not equals) are: location, resourceType, name, resourceGroup, 398 identity, identity/principalId, plan, plan/publisher, plan/product, plan/name, plan/version, 399 and plan/promotionCode.:code:`<br>`:code:`<br>`For example, to filter by a resource type, use: 400 $filter=resourceType eq 'Microsoft.Network/virtualNetworks':code:`<br>`:code:`<br>`You can use 401 substringof(value, property) in the filter. The properties you can use for substring are: name 402 and resourceGroup.:code:`<br>`:code:`<br>`For example, to get all resources with 'demo' 403 anywhere in the name, use: $filter=substringof('demo', name):code:`<br>`:code:`<br>`You can 404 link more than one substringof together by adding and/or operators.:code:`<br>`:code:`<br>`You 405 can filter by tag names and values. For example, to filter for a tag name and value, use 406 $filter=tagName eq 'tag1' and tagValue eq 'Value1':code:`<br>`:code:`<br>`You can use some 407 properties together when filtering. The combinations you can use are: substringof and/or 408 resourceType, plan and plan/publisher and plan/name, identity and identity/principalId. 409 :type filter: str 410 :param expand: Comma-separated list of additional properties to be included in the response. 411 Valid values include ``createdTime``\ , ``changedTime`` and ``provisioningState``. For example, 412 ``$expand=createdTime,changedTime``. 413 :type expand: str 414 :param top: The number of results to return. If null is passed, returns all resource groups. 415 :type top: int 416 :keyword callable cls: A custom type or function that will be passed the direct response 417 :return: An iterator like instance of either ResourceListResult or the result of cls(response) 418 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2019_05_01.models.ResourceListResult] 419 :raises: ~azure.core.exceptions.HttpResponseError 420 """ 421 cls = kwargs.pop('cls', None) # type: ClsType["_models.ResourceListResult"] 422 error_map = { 423 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 424 } 425 error_map.update(kwargs.pop('error_map', {})) 426 api_version = "2019-05-01" 427 accept = "application/json" 428 429 def prepare_request(next_link=None): 430 # Construct headers 431 header_parameters = {} # type: Dict[str, Any] 432 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 433 434 if not next_link: 435 # Construct URL 436 url = self.list.metadata['url'] # type: ignore 437 path_format_arguments = { 438 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 439 } 440 url = self._client.format_url(url, **path_format_arguments) 441 # Construct parameters 442 query_parameters = {} # type: Dict[str, Any] 443 if filter is not None: 444 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str') 445 if expand is not None: 446 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 447 if top is not None: 448 query_parameters['$top'] = self._serialize.query("top", top, 'int') 449 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 450 451 request = self._client.get(url, query_parameters, header_parameters) 452 else: 453 url = next_link 454 query_parameters = {} # type: Dict[str, Any] 455 request = self._client.get(url, query_parameters, header_parameters) 456 return request 457 458 def extract_data(pipeline_response): 459 deserialized = self._deserialize('ResourceListResult', pipeline_response) 460 list_of_elem = deserialized.value 461 if cls: 462 list_of_elem = cls(list_of_elem) 463 return deserialized.next_link or None, iter(list_of_elem) 464 465 def get_next(next_link=None): 466 request = prepare_request(next_link) 467 468 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 469 response = pipeline_response.http_response 470 471 if response.status_code not in [200]: 472 map_error(status_code=response.status_code, response=response, error_map=error_map) 473 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 474 475 return pipeline_response 476 477 return ItemPaged( 478 get_next, extract_data 479 ) 480 list.metadata = {'url': '/subscriptions/{subscriptionId}/resources'} # type: ignore 481 482 def check_existence( 483 self, 484 resource_group_name, # type: str 485 resource_provider_namespace, # type: str 486 parent_resource_path, # type: str 487 resource_type, # type: str 488 resource_name, # type: str 489 api_version, # type: str 490 **kwargs # type: Any 491 ): 492 # type: (...) -> bool 493 """Checks whether a resource exists. 494 495 :param resource_group_name: The name of the resource group containing the resource to check. 496 The name is case insensitive. 497 :type resource_group_name: str 498 :param resource_provider_namespace: The resource provider of the resource to check. 499 :type resource_provider_namespace: str 500 :param parent_resource_path: The parent resource identity. 501 :type parent_resource_path: str 502 :param resource_type: The resource type. 503 :type resource_type: str 504 :param resource_name: The name of the resource to check whether it exists. 505 :type resource_name: str 506 :param api_version: The API version to use for the operation. 507 :type api_version: str 508 :keyword callable cls: A custom type or function that will be passed the direct response 509 :return: bool, or the result of cls(response) 510 :rtype: bool 511 :raises: ~azure.core.exceptions.HttpResponseError 512 """ 513 cls = kwargs.pop('cls', None) # type: ClsType[None] 514 error_map = { 515 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 516 } 517 error_map.update(kwargs.pop('error_map', {})) 518 519 # Construct URL 520 url = self.check_existence.metadata['url'] # type: ignore 521 path_format_arguments = { 522 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 523 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 524 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 525 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 526 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 527 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 528 } 529 url = self._client.format_url(url, **path_format_arguments) 530 531 # Construct parameters 532 query_parameters = {} # type: Dict[str, Any] 533 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 534 535 # Construct headers 536 header_parameters = {} # type: Dict[str, Any] 537 538 request = self._client.head(url, query_parameters, header_parameters) 539 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 540 response = pipeline_response.http_response 541 542 if response.status_code not in [204, 404]: 543 map_error(status_code=response.status_code, response=response, error_map=error_map) 544 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 545 546 if cls: 547 return cls(pipeline_response, None, {}) 548 549 return 200 <= response.status_code <= 299 550 check_existence.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 551 552 def _delete_initial( 553 self, 554 resource_group_name, # type: str 555 resource_provider_namespace, # type: str 556 parent_resource_path, # type: str 557 resource_type, # type: str 558 resource_name, # type: str 559 api_version, # type: str 560 **kwargs # type: Any 561 ): 562 # type: (...) -> None 563 cls = kwargs.pop('cls', None) # type: ClsType[None] 564 error_map = { 565 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 566 } 567 error_map.update(kwargs.pop('error_map', {})) 568 569 # Construct URL 570 url = self._delete_initial.metadata['url'] # type: ignore 571 path_format_arguments = { 572 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 573 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 574 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 575 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 576 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 577 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 578 } 579 url = self._client.format_url(url, **path_format_arguments) 580 581 # Construct parameters 582 query_parameters = {} # type: Dict[str, Any] 583 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 584 585 # Construct headers 586 header_parameters = {} # type: Dict[str, Any] 587 588 request = self._client.delete(url, query_parameters, header_parameters) 589 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 590 response = pipeline_response.http_response 591 592 if response.status_code not in [200, 202, 204]: 593 map_error(status_code=response.status_code, response=response, error_map=error_map) 594 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 595 596 if cls: 597 return cls(pipeline_response, None, {}) 598 599 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 600 601 def begin_delete( 602 self, 603 resource_group_name, # type: str 604 resource_provider_namespace, # type: str 605 parent_resource_path, # type: str 606 resource_type, # type: str 607 resource_name, # type: str 608 api_version, # type: str 609 **kwargs # type: Any 610 ): 611 # type: (...) -> LROPoller[None] 612 """Deletes a resource. 613 614 :param resource_group_name: The name of the resource group that contains the resource to 615 delete. The name is case insensitive. 616 :type resource_group_name: str 617 :param resource_provider_namespace: The namespace of the resource provider. 618 :type resource_provider_namespace: str 619 :param parent_resource_path: The parent resource identity. 620 :type parent_resource_path: str 621 :param resource_type: The resource type. 622 :type resource_type: str 623 :param resource_name: The name of the resource to delete. 624 :type resource_name: str 625 :param api_version: The API version to use for the operation. 626 :type api_version: str 627 :keyword callable cls: A custom type or function that will be passed the direct response 628 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 629 :keyword polling: By default, your polling method will be ARMPolling. 630 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 631 :paramtype polling: bool or ~azure.core.polling.PollingMethod 632 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 633 :return: An instance of LROPoller that returns either None or the result of cls(response) 634 :rtype: ~azure.core.polling.LROPoller[None] 635 :raises ~azure.core.exceptions.HttpResponseError: 636 """ 637 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 638 cls = kwargs.pop('cls', None) # type: ClsType[None] 639 lro_delay = kwargs.pop( 640 'polling_interval', 641 self._config.polling_interval 642 ) 643 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 644 if cont_token is None: 645 raw_result = self._delete_initial( 646 resource_group_name=resource_group_name, 647 resource_provider_namespace=resource_provider_namespace, 648 parent_resource_path=parent_resource_path, 649 resource_type=resource_type, 650 resource_name=resource_name, 651 api_version=api_version, 652 cls=lambda x,y,z: x, 653 **kwargs 654 ) 655 656 kwargs.pop('error_map', None) 657 kwargs.pop('content_type', None) 658 659 def get_long_running_output(pipeline_response): 660 if cls: 661 return cls(pipeline_response, None, {}) 662 663 path_format_arguments = { 664 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 665 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 666 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 667 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 668 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 669 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 670 } 671 672 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 673 elif polling is False: polling_method = NoPolling() 674 else: polling_method = polling 675 if cont_token: 676 return LROPoller.from_continuation_token( 677 polling_method=polling_method, 678 continuation_token=cont_token, 679 client=self._client, 680 deserialization_callback=get_long_running_output 681 ) 682 else: 683 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 684 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 685 686 def _create_or_update_initial( 687 self, 688 resource_group_name, # type: str 689 resource_provider_namespace, # type: str 690 parent_resource_path, # type: str 691 resource_type, # type: str 692 resource_name, # type: str 693 api_version, # type: str 694 parameters, # type: "_models.GenericResource" 695 **kwargs # type: Any 696 ): 697 # type: (...) -> Optional["_models.GenericResource"] 698 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.GenericResource"]] 699 error_map = { 700 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 701 } 702 error_map.update(kwargs.pop('error_map', {})) 703 content_type = kwargs.pop("content_type", "application/json") 704 accept = "application/json" 705 706 # Construct URL 707 url = self._create_or_update_initial.metadata['url'] # type: ignore 708 path_format_arguments = { 709 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 710 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 711 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 712 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 713 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 714 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 715 } 716 url = self._client.format_url(url, **path_format_arguments) 717 718 # Construct parameters 719 query_parameters = {} # type: Dict[str, Any] 720 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 721 722 # Construct headers 723 header_parameters = {} # type: Dict[str, Any] 724 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 725 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 726 727 body_content_kwargs = {} # type: Dict[str, Any] 728 body_content = self._serialize.body(parameters, 'GenericResource') 729 body_content_kwargs['content'] = body_content 730 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 731 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 732 response = pipeline_response.http_response 733 734 if response.status_code not in [200, 201, 202]: 735 map_error(status_code=response.status_code, response=response, error_map=error_map) 736 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 737 738 deserialized = None 739 if response.status_code == 200: 740 deserialized = self._deserialize('GenericResource', pipeline_response) 741 742 if response.status_code == 201: 743 deserialized = self._deserialize('GenericResource', pipeline_response) 744 745 if cls: 746 return cls(pipeline_response, deserialized, {}) 747 748 return deserialized 749 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 750 751 def begin_create_or_update( 752 self, 753 resource_group_name, # type: str 754 resource_provider_namespace, # type: str 755 parent_resource_path, # type: str 756 resource_type, # type: str 757 resource_name, # type: str 758 api_version, # type: str 759 parameters, # type: "_models.GenericResource" 760 **kwargs # type: Any 761 ): 762 # type: (...) -> LROPoller["_models.GenericResource"] 763 """Creates a resource. 764 765 :param resource_group_name: The name of the resource group for the resource. The name is case 766 insensitive. 767 :type resource_group_name: str 768 :param resource_provider_namespace: The namespace of the resource provider. 769 :type resource_provider_namespace: str 770 :param parent_resource_path: The parent resource identity. 771 :type parent_resource_path: str 772 :param resource_type: The resource type of the resource to create. 773 :type resource_type: str 774 :param resource_name: The name of the resource to create. 775 :type resource_name: str 776 :param api_version: The API version to use for the operation. 777 :type api_version: str 778 :param parameters: Parameters for creating or updating the resource. 779 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 780 :keyword callable cls: A custom type or function that will be passed the direct response 781 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 782 :keyword polling: By default, your polling method will be ARMPolling. 783 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 784 :paramtype polling: bool or ~azure.core.polling.PollingMethod 785 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 786 :return: An instance of LROPoller that returns either GenericResource or the result of cls(response) 787 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource] 788 :raises ~azure.core.exceptions.HttpResponseError: 789 """ 790 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 791 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 792 lro_delay = kwargs.pop( 793 'polling_interval', 794 self._config.polling_interval 795 ) 796 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 797 if cont_token is None: 798 raw_result = self._create_or_update_initial( 799 resource_group_name=resource_group_name, 800 resource_provider_namespace=resource_provider_namespace, 801 parent_resource_path=parent_resource_path, 802 resource_type=resource_type, 803 resource_name=resource_name, 804 api_version=api_version, 805 parameters=parameters, 806 cls=lambda x,y,z: x, 807 **kwargs 808 ) 809 810 kwargs.pop('error_map', None) 811 kwargs.pop('content_type', None) 812 813 def get_long_running_output(pipeline_response): 814 deserialized = self._deserialize('GenericResource', pipeline_response) 815 816 if cls: 817 return cls(pipeline_response, deserialized, {}) 818 return deserialized 819 820 path_format_arguments = { 821 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 822 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 823 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 824 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 825 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 826 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 827 } 828 829 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 830 elif polling is False: polling_method = NoPolling() 831 else: polling_method = polling 832 if cont_token: 833 return LROPoller.from_continuation_token( 834 polling_method=polling_method, 835 continuation_token=cont_token, 836 client=self._client, 837 deserialization_callback=get_long_running_output 838 ) 839 else: 840 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 841 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 842 843 def _update_initial( 844 self, 845 resource_group_name, # type: str 846 resource_provider_namespace, # type: str 847 parent_resource_path, # type: str 848 resource_type, # type: str 849 resource_name, # type: str 850 api_version, # type: str 851 parameters, # type: "_models.GenericResource" 852 **kwargs # type: Any 853 ): 854 # type: (...) -> Optional["_models.GenericResource"] 855 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.GenericResource"]] 856 error_map = { 857 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 858 } 859 error_map.update(kwargs.pop('error_map', {})) 860 content_type = kwargs.pop("content_type", "application/json") 861 accept = "application/json" 862 863 # Construct URL 864 url = self._update_initial.metadata['url'] # type: ignore 865 path_format_arguments = { 866 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 867 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 868 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 869 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 870 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 871 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 872 } 873 url = self._client.format_url(url, **path_format_arguments) 874 875 # Construct parameters 876 query_parameters = {} # type: Dict[str, Any] 877 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 878 879 # Construct headers 880 header_parameters = {} # type: Dict[str, Any] 881 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 882 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 883 884 body_content_kwargs = {} # type: Dict[str, Any] 885 body_content = self._serialize.body(parameters, 'GenericResource') 886 body_content_kwargs['content'] = body_content 887 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 888 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 889 response = pipeline_response.http_response 890 891 if response.status_code not in [200, 202]: 892 map_error(status_code=response.status_code, response=response, error_map=error_map) 893 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 894 895 deserialized = None 896 if response.status_code == 200: 897 deserialized = self._deserialize('GenericResource', pipeline_response) 898 899 if cls: 900 return cls(pipeline_response, deserialized, {}) 901 902 return deserialized 903 _update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 904 905 def begin_update( 906 self, 907 resource_group_name, # type: str 908 resource_provider_namespace, # type: str 909 parent_resource_path, # type: str 910 resource_type, # type: str 911 resource_name, # type: str 912 api_version, # type: str 913 parameters, # type: "_models.GenericResource" 914 **kwargs # type: Any 915 ): 916 # type: (...) -> LROPoller["_models.GenericResource"] 917 """Updates a resource. 918 919 :param resource_group_name: The name of the resource group for the resource. The name is case 920 insensitive. 921 :type resource_group_name: str 922 :param resource_provider_namespace: The namespace of the resource provider. 923 :type resource_provider_namespace: str 924 :param parent_resource_path: The parent resource identity. 925 :type parent_resource_path: str 926 :param resource_type: The resource type of the resource to update. 927 :type resource_type: str 928 :param resource_name: The name of the resource to update. 929 :type resource_name: str 930 :param api_version: The API version to use for the operation. 931 :type api_version: str 932 :param parameters: Parameters for updating the resource. 933 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 934 :keyword callable cls: A custom type or function that will be passed the direct response 935 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 936 :keyword polling: By default, your polling method will be ARMPolling. 937 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 938 :paramtype polling: bool or ~azure.core.polling.PollingMethod 939 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 940 :return: An instance of LROPoller that returns either GenericResource or the result of cls(response) 941 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource] 942 :raises ~azure.core.exceptions.HttpResponseError: 943 """ 944 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 945 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 946 lro_delay = kwargs.pop( 947 'polling_interval', 948 self._config.polling_interval 949 ) 950 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 951 if cont_token is None: 952 raw_result = self._update_initial( 953 resource_group_name=resource_group_name, 954 resource_provider_namespace=resource_provider_namespace, 955 parent_resource_path=parent_resource_path, 956 resource_type=resource_type, 957 resource_name=resource_name, 958 api_version=api_version, 959 parameters=parameters, 960 cls=lambda x,y,z: x, 961 **kwargs 962 ) 963 964 kwargs.pop('error_map', None) 965 kwargs.pop('content_type', None) 966 967 def get_long_running_output(pipeline_response): 968 deserialized = self._deserialize('GenericResource', pipeline_response) 969 970 if cls: 971 return cls(pipeline_response, deserialized, {}) 972 return deserialized 973 974 path_format_arguments = { 975 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 976 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 977 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 978 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 979 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 980 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 981 } 982 983 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 984 elif polling is False: polling_method = NoPolling() 985 else: polling_method = polling 986 if cont_token: 987 return LROPoller.from_continuation_token( 988 polling_method=polling_method, 989 continuation_token=cont_token, 990 client=self._client, 991 deserialization_callback=get_long_running_output 992 ) 993 else: 994 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 995 begin_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 996 997 def get( 998 self, 999 resource_group_name, # type: str 1000 resource_provider_namespace, # type: str 1001 parent_resource_path, # type: str 1002 resource_type, # type: str 1003 resource_name, # type: str 1004 api_version, # type: str 1005 **kwargs # type: Any 1006 ): 1007 # type: (...) -> "_models.GenericResource" 1008 """Gets a resource. 1009 1010 :param resource_group_name: The name of the resource group containing the resource to get. The 1011 name is case insensitive. 1012 :type resource_group_name: str 1013 :param resource_provider_namespace: The namespace of the resource provider. 1014 :type resource_provider_namespace: str 1015 :param parent_resource_path: The parent resource identity. 1016 :type parent_resource_path: str 1017 :param resource_type: The resource type of the resource. 1018 :type resource_type: str 1019 :param resource_name: The name of the resource to get. 1020 :type resource_name: str 1021 :param api_version: The API version to use for the operation. 1022 :type api_version: str 1023 :keyword callable cls: A custom type or function that will be passed the direct response 1024 :return: GenericResource, or the result of cls(response) 1025 :rtype: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 1026 :raises: ~azure.core.exceptions.HttpResponseError 1027 """ 1028 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 1029 error_map = { 1030 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1031 } 1032 error_map.update(kwargs.pop('error_map', {})) 1033 accept = "application/json" 1034 1035 # Construct URL 1036 url = self.get.metadata['url'] # type: ignore 1037 path_format_arguments = { 1038 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 1039 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 1040 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 1041 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 1042 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 1043 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1044 } 1045 url = self._client.format_url(url, **path_format_arguments) 1046 1047 # Construct parameters 1048 query_parameters = {} # type: Dict[str, Any] 1049 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1050 1051 # Construct headers 1052 header_parameters = {} # type: Dict[str, Any] 1053 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1054 1055 request = self._client.get(url, query_parameters, header_parameters) 1056 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1057 response = pipeline_response.http_response 1058 1059 if response.status_code not in [200]: 1060 map_error(status_code=response.status_code, response=response, error_map=error_map) 1061 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1062 1063 deserialized = self._deserialize('GenericResource', pipeline_response) 1064 1065 if cls: 1066 return cls(pipeline_response, deserialized, {}) 1067 1068 return deserialized 1069 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}'} # type: ignore 1070 1071 def check_existence_by_id( 1072 self, 1073 resource_id, # type: str 1074 api_version, # type: str 1075 **kwargs # type: Any 1076 ): 1077 # type: (...) -> bool 1078 """Checks by ID whether a resource exists. 1079 1080 :param resource_id: The fully qualified ID of the resource, including the resource name and 1081 resource type. Use the format, 1082 /subscriptions/{guid}/resourceGroups/{resource-group-name}/{resource-provider-namespace}/{resource-type}/{resource-name}. 1083 :type resource_id: str 1084 :param api_version: The API version to use for the operation. 1085 :type api_version: str 1086 :keyword callable cls: A custom type or function that will be passed the direct response 1087 :return: bool, or the result of cls(response) 1088 :rtype: bool 1089 :raises: ~azure.core.exceptions.HttpResponseError 1090 """ 1091 cls = kwargs.pop('cls', None) # type: ClsType[None] 1092 error_map = { 1093 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1094 } 1095 error_map.update(kwargs.pop('error_map', {})) 1096 1097 # Construct URL 1098 url = self.check_existence_by_id.metadata['url'] # type: ignore 1099 path_format_arguments = { 1100 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1101 } 1102 url = self._client.format_url(url, **path_format_arguments) 1103 1104 # Construct parameters 1105 query_parameters = {} # type: Dict[str, Any] 1106 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1107 1108 # Construct headers 1109 header_parameters = {} # type: Dict[str, Any] 1110 1111 request = self._client.head(url, query_parameters, header_parameters) 1112 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1113 response = pipeline_response.http_response 1114 1115 if response.status_code not in [204, 404]: 1116 map_error(status_code=response.status_code, response=response, error_map=error_map) 1117 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1118 1119 if cls: 1120 return cls(pipeline_response, None, {}) 1121 1122 return 200 <= response.status_code <= 299 1123 check_existence_by_id.metadata = {'url': '/{resourceId}'} # type: ignore 1124 1125 def _delete_by_id_initial( 1126 self, 1127 resource_id, # type: str 1128 api_version, # type: str 1129 **kwargs # type: Any 1130 ): 1131 # type: (...) -> None 1132 cls = kwargs.pop('cls', None) # type: ClsType[None] 1133 error_map = { 1134 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1135 } 1136 error_map.update(kwargs.pop('error_map', {})) 1137 1138 # Construct URL 1139 url = self._delete_by_id_initial.metadata['url'] # type: ignore 1140 path_format_arguments = { 1141 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1142 } 1143 url = self._client.format_url(url, **path_format_arguments) 1144 1145 # Construct parameters 1146 query_parameters = {} # type: Dict[str, Any] 1147 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1148 1149 # Construct headers 1150 header_parameters = {} # type: Dict[str, Any] 1151 1152 request = self._client.delete(url, query_parameters, header_parameters) 1153 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1154 response = pipeline_response.http_response 1155 1156 if response.status_code not in [200, 202, 204]: 1157 map_error(status_code=response.status_code, response=response, error_map=error_map) 1158 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1159 1160 if cls: 1161 return cls(pipeline_response, None, {}) 1162 1163 _delete_by_id_initial.metadata = {'url': '/{resourceId}'} # type: ignore 1164 1165 def begin_delete_by_id( 1166 self, 1167 resource_id, # type: str 1168 api_version, # type: str 1169 **kwargs # type: Any 1170 ): 1171 # type: (...) -> LROPoller[None] 1172 """Deletes a resource by ID. 1173 1174 :param resource_id: The fully qualified ID of the resource, including the resource name and 1175 resource type. Use the format, 1176 /subscriptions/{guid}/resourceGroups/{resource-group-name}/{resource-provider-namespace}/{resource-type}/{resource-name}. 1177 :type resource_id: str 1178 :param api_version: The API version to use for the operation. 1179 :type api_version: str 1180 :keyword callable cls: A custom type or function that will be passed the direct response 1181 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 1182 :keyword polling: By default, your polling method will be ARMPolling. 1183 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 1184 :paramtype polling: bool or ~azure.core.polling.PollingMethod 1185 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 1186 :return: An instance of LROPoller that returns either None or the result of cls(response) 1187 :rtype: ~azure.core.polling.LROPoller[None] 1188 :raises ~azure.core.exceptions.HttpResponseError: 1189 """ 1190 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 1191 cls = kwargs.pop('cls', None) # type: ClsType[None] 1192 lro_delay = kwargs.pop( 1193 'polling_interval', 1194 self._config.polling_interval 1195 ) 1196 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 1197 if cont_token is None: 1198 raw_result = self._delete_by_id_initial( 1199 resource_id=resource_id, 1200 api_version=api_version, 1201 cls=lambda x,y,z: x, 1202 **kwargs 1203 ) 1204 1205 kwargs.pop('error_map', None) 1206 kwargs.pop('content_type', None) 1207 1208 def get_long_running_output(pipeline_response): 1209 if cls: 1210 return cls(pipeline_response, None, {}) 1211 1212 path_format_arguments = { 1213 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1214 } 1215 1216 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 1217 elif polling is False: polling_method = NoPolling() 1218 else: polling_method = polling 1219 if cont_token: 1220 return LROPoller.from_continuation_token( 1221 polling_method=polling_method, 1222 continuation_token=cont_token, 1223 client=self._client, 1224 deserialization_callback=get_long_running_output 1225 ) 1226 else: 1227 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 1228 begin_delete_by_id.metadata = {'url': '/{resourceId}'} # type: ignore 1229 1230 def _create_or_update_by_id_initial( 1231 self, 1232 resource_id, # type: str 1233 api_version, # type: str 1234 parameters, # type: "_models.GenericResource" 1235 **kwargs # type: Any 1236 ): 1237 # type: (...) -> Optional["_models.GenericResource"] 1238 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.GenericResource"]] 1239 error_map = { 1240 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1241 } 1242 error_map.update(kwargs.pop('error_map', {})) 1243 content_type = kwargs.pop("content_type", "application/json") 1244 accept = "application/json" 1245 1246 # Construct URL 1247 url = self._create_or_update_by_id_initial.metadata['url'] # type: ignore 1248 path_format_arguments = { 1249 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1250 } 1251 url = self._client.format_url(url, **path_format_arguments) 1252 1253 # Construct parameters 1254 query_parameters = {} # type: Dict[str, Any] 1255 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1256 1257 # Construct headers 1258 header_parameters = {} # type: Dict[str, Any] 1259 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 1260 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1261 1262 body_content_kwargs = {} # type: Dict[str, Any] 1263 body_content = self._serialize.body(parameters, 'GenericResource') 1264 body_content_kwargs['content'] = body_content 1265 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 1266 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1267 response = pipeline_response.http_response 1268 1269 if response.status_code not in [200, 201, 202]: 1270 map_error(status_code=response.status_code, response=response, error_map=error_map) 1271 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1272 1273 deserialized = None 1274 if response.status_code == 200: 1275 deserialized = self._deserialize('GenericResource', pipeline_response) 1276 1277 if response.status_code == 201: 1278 deserialized = self._deserialize('GenericResource', pipeline_response) 1279 1280 if cls: 1281 return cls(pipeline_response, deserialized, {}) 1282 1283 return deserialized 1284 _create_or_update_by_id_initial.metadata = {'url': '/{resourceId}'} # type: ignore 1285 1286 def begin_create_or_update_by_id( 1287 self, 1288 resource_id, # type: str 1289 api_version, # type: str 1290 parameters, # type: "_models.GenericResource" 1291 **kwargs # type: Any 1292 ): 1293 # type: (...) -> LROPoller["_models.GenericResource"] 1294 """Create a resource by ID. 1295 1296 :param resource_id: The fully qualified ID of the resource, including the resource name and 1297 resource type. Use the format, 1298 /subscriptions/{guid}/resourceGroups/{resource-group-name}/{resource-provider-namespace}/{resource-type}/{resource-name}. 1299 :type resource_id: str 1300 :param api_version: The API version to use for the operation. 1301 :type api_version: str 1302 :param parameters: Create or update resource parameters. 1303 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 1304 :keyword callable cls: A custom type or function that will be passed the direct response 1305 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 1306 :keyword polling: By default, your polling method will be ARMPolling. 1307 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 1308 :paramtype polling: bool or ~azure.core.polling.PollingMethod 1309 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 1310 :return: An instance of LROPoller that returns either GenericResource or the result of cls(response) 1311 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource] 1312 :raises ~azure.core.exceptions.HttpResponseError: 1313 """ 1314 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 1315 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 1316 lro_delay = kwargs.pop( 1317 'polling_interval', 1318 self._config.polling_interval 1319 ) 1320 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 1321 if cont_token is None: 1322 raw_result = self._create_or_update_by_id_initial( 1323 resource_id=resource_id, 1324 api_version=api_version, 1325 parameters=parameters, 1326 cls=lambda x,y,z: x, 1327 **kwargs 1328 ) 1329 1330 kwargs.pop('error_map', None) 1331 kwargs.pop('content_type', None) 1332 1333 def get_long_running_output(pipeline_response): 1334 deserialized = self._deserialize('GenericResource', pipeline_response) 1335 1336 if cls: 1337 return cls(pipeline_response, deserialized, {}) 1338 return deserialized 1339 1340 path_format_arguments = { 1341 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1342 } 1343 1344 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 1345 elif polling is False: polling_method = NoPolling() 1346 else: polling_method = polling 1347 if cont_token: 1348 return LROPoller.from_continuation_token( 1349 polling_method=polling_method, 1350 continuation_token=cont_token, 1351 client=self._client, 1352 deserialization_callback=get_long_running_output 1353 ) 1354 else: 1355 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 1356 begin_create_or_update_by_id.metadata = {'url': '/{resourceId}'} # type: ignore 1357 1358 def _update_by_id_initial( 1359 self, 1360 resource_id, # type: str 1361 api_version, # type: str 1362 parameters, # type: "_models.GenericResource" 1363 **kwargs # type: Any 1364 ): 1365 # type: (...) -> Optional["_models.GenericResource"] 1366 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.GenericResource"]] 1367 error_map = { 1368 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1369 } 1370 error_map.update(kwargs.pop('error_map', {})) 1371 content_type = kwargs.pop("content_type", "application/json") 1372 accept = "application/json" 1373 1374 # Construct URL 1375 url = self._update_by_id_initial.metadata['url'] # type: ignore 1376 path_format_arguments = { 1377 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1378 } 1379 url = self._client.format_url(url, **path_format_arguments) 1380 1381 # Construct parameters 1382 query_parameters = {} # type: Dict[str, Any] 1383 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1384 1385 # Construct headers 1386 header_parameters = {} # type: Dict[str, Any] 1387 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 1388 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1389 1390 body_content_kwargs = {} # type: Dict[str, Any] 1391 body_content = self._serialize.body(parameters, 'GenericResource') 1392 body_content_kwargs['content'] = body_content 1393 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 1394 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1395 response = pipeline_response.http_response 1396 1397 if response.status_code not in [200, 202]: 1398 map_error(status_code=response.status_code, response=response, error_map=error_map) 1399 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1400 1401 deserialized = None 1402 if response.status_code == 200: 1403 deserialized = self._deserialize('GenericResource', pipeline_response) 1404 1405 if cls: 1406 return cls(pipeline_response, deserialized, {}) 1407 1408 return deserialized 1409 _update_by_id_initial.metadata = {'url': '/{resourceId}'} # type: ignore 1410 1411 def begin_update_by_id( 1412 self, 1413 resource_id, # type: str 1414 api_version, # type: str 1415 parameters, # type: "_models.GenericResource" 1416 **kwargs # type: Any 1417 ): 1418 # type: (...) -> LROPoller["_models.GenericResource"] 1419 """Updates a resource by ID. 1420 1421 :param resource_id: The fully qualified ID of the resource, including the resource name and 1422 resource type. Use the format, 1423 /subscriptions/{guid}/resourceGroups/{resource-group-name}/{resource-provider-namespace}/{resource-type}/{resource-name}. 1424 :type resource_id: str 1425 :param api_version: The API version to use for the operation. 1426 :type api_version: str 1427 :param parameters: Update resource parameters. 1428 :type parameters: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 1429 :keyword callable cls: A custom type or function that will be passed the direct response 1430 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 1431 :keyword polling: By default, your polling method will be ARMPolling. 1432 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 1433 :paramtype polling: bool or ~azure.core.polling.PollingMethod 1434 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 1435 :return: An instance of LROPoller that returns either GenericResource or the result of cls(response) 1436 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource] 1437 :raises ~azure.core.exceptions.HttpResponseError: 1438 """ 1439 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 1440 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 1441 lro_delay = kwargs.pop( 1442 'polling_interval', 1443 self._config.polling_interval 1444 ) 1445 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 1446 if cont_token is None: 1447 raw_result = self._update_by_id_initial( 1448 resource_id=resource_id, 1449 api_version=api_version, 1450 parameters=parameters, 1451 cls=lambda x,y,z: x, 1452 **kwargs 1453 ) 1454 1455 kwargs.pop('error_map', None) 1456 kwargs.pop('content_type', None) 1457 1458 def get_long_running_output(pipeline_response): 1459 deserialized = self._deserialize('GenericResource', pipeline_response) 1460 1461 if cls: 1462 return cls(pipeline_response, deserialized, {}) 1463 return deserialized 1464 1465 path_format_arguments = { 1466 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1467 } 1468 1469 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 1470 elif polling is False: polling_method = NoPolling() 1471 else: polling_method = polling 1472 if cont_token: 1473 return LROPoller.from_continuation_token( 1474 polling_method=polling_method, 1475 continuation_token=cont_token, 1476 client=self._client, 1477 deserialization_callback=get_long_running_output 1478 ) 1479 else: 1480 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 1481 begin_update_by_id.metadata = {'url': '/{resourceId}'} # type: ignore 1482 1483 def get_by_id( 1484 self, 1485 resource_id, # type: str 1486 api_version, # type: str 1487 **kwargs # type: Any 1488 ): 1489 # type: (...) -> "_models.GenericResource" 1490 """Gets a resource by ID. 1491 1492 :param resource_id: The fully qualified ID of the resource, including the resource name and 1493 resource type. Use the format, 1494 /subscriptions/{guid}/resourceGroups/{resource-group-name}/{resource-provider-namespace}/{resource-type}/{resource-name}. 1495 :type resource_id: str 1496 :param api_version: The API version to use for the operation. 1497 :type api_version: str 1498 :keyword callable cls: A custom type or function that will be passed the direct response 1499 :return: GenericResource, or the result of cls(response) 1500 :rtype: ~azure.mgmt.resource.resources.v2019_05_01.models.GenericResource 1501 :raises: ~azure.core.exceptions.HttpResponseError 1502 """ 1503 cls = kwargs.pop('cls', None) # type: ClsType["_models.GenericResource"] 1504 error_map = { 1505 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1506 } 1507 error_map.update(kwargs.pop('error_map', {})) 1508 accept = "application/json" 1509 1510 # Construct URL 1511 url = self.get_by_id.metadata['url'] # type: ignore 1512 path_format_arguments = { 1513 'resourceId': self._serialize.url("resource_id", resource_id, 'str', skip_quote=True), 1514 } 1515 url = self._client.format_url(url, **path_format_arguments) 1516 1517 # Construct parameters 1518 query_parameters = {} # type: Dict[str, Any] 1519 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1520 1521 # Construct headers 1522 header_parameters = {} # type: Dict[str, Any] 1523 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1524 1525 request = self._client.get(url, query_parameters, header_parameters) 1526 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1527 response = pipeline_response.http_response 1528 1529 if response.status_code not in [200]: 1530 map_error(status_code=response.status_code, response=response, error_map=error_map) 1531 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1532 1533 deserialized = self._deserialize('GenericResource', pipeline_response) 1534 1535 if cls: 1536 return cls(pipeline_response, deserialized, {}) 1537 1538 return deserialized 1539 get_by_id.metadata = {'url': '/{resourceId}'} # type: ignore 1540