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.mgmt.core.exceptions import ARMErrorFormat 16 17from .. import models as _models 18 19if TYPE_CHECKING: 20 # pylint: disable=unused-import,ungrouped-imports 21 from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar 22 23 T = TypeVar('T') 24 ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] 25 26class FeaturesOperations(object): 27 """FeaturesOperations operations. 28 29 You should not instantiate this class directly. Instead, you should create a Client instance that 30 instantiates it for you and attaches it as an attribute. 31 32 :ivar models: Alias to model classes used in this operation group. 33 :type models: ~azure.mgmt.resource.features.v2021_07_01.models 34 :param client: Client for service requests. 35 :param config: Configuration of service client. 36 :param serializer: An object model serializer. 37 :param deserializer: An object model deserializer. 38 """ 39 40 models = _models 41 42 def __init__(self, client, config, serializer, deserializer): 43 self._client = client 44 self._serialize = serializer 45 self._deserialize = deserializer 46 self._config = config 47 48 def list_all( 49 self, 50 **kwargs # type: Any 51 ): 52 # type: (...) -> Iterable["_models.FeatureOperationsListResult"] 53 """Gets all the preview features that are available through AFEC for the subscription. 54 55 :keyword callable cls: A custom type or function that will be passed the direct response 56 :return: An iterator like instance of either FeatureOperationsListResult or the result of cls(response) 57 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.features.v2021_07_01.models.FeatureOperationsListResult] 58 :raises: ~azure.core.exceptions.HttpResponseError 59 """ 60 cls = kwargs.pop('cls', None) # type: ClsType["_models.FeatureOperationsListResult"] 61 error_map = { 62 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 63 } 64 error_map.update(kwargs.pop('error_map', {})) 65 api_version = "2021-07-01" 66 accept = "application/json, text/json" 67 68 def prepare_request(next_link=None): 69 # Construct headers 70 header_parameters = {} # type: Dict[str, Any] 71 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 72 73 if not next_link: 74 # Construct URL 75 url = self.list_all.metadata['url'] # type: ignore 76 path_format_arguments = { 77 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 78 } 79 url = self._client.format_url(url, **path_format_arguments) 80 # Construct parameters 81 query_parameters = {} # type: Dict[str, Any] 82 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 83 84 request = self._client.get(url, query_parameters, header_parameters) 85 else: 86 url = next_link 87 query_parameters = {} # type: Dict[str, Any] 88 request = self._client.get(url, query_parameters, header_parameters) 89 return request 90 91 def extract_data(pipeline_response): 92 deserialized = self._deserialize('FeatureOperationsListResult', pipeline_response) 93 list_of_elem = deserialized.value 94 if cls: 95 list_of_elem = cls(list_of_elem) 96 return deserialized.next_link or None, iter(list_of_elem) 97 98 def get_next(next_link=None): 99 request = prepare_request(next_link) 100 101 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 102 response = pipeline_response.http_response 103 104 if response.status_code not in [200]: 105 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 106 map_error(status_code=response.status_code, response=response, error_map=error_map) 107 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 108 109 return pipeline_response 110 111 return ItemPaged( 112 get_next, extract_data 113 ) 114 list_all.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Features/features'} # type: ignore 115 116 def list( 117 self, 118 resource_provider_namespace, # type: str 119 **kwargs # type: Any 120 ): 121 # type: (...) -> Iterable["_models.FeatureOperationsListResult"] 122 """Gets all the preview features in a provider namespace that are available through AFEC for the 123 subscription. 124 125 :param resource_provider_namespace: The namespace of the resource provider for getting 126 features. 127 :type resource_provider_namespace: str 128 :keyword callable cls: A custom type or function that will be passed the direct response 129 :return: An iterator like instance of either FeatureOperationsListResult or the result of cls(response) 130 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.features.v2021_07_01.models.FeatureOperationsListResult] 131 :raises: ~azure.core.exceptions.HttpResponseError 132 """ 133 cls = kwargs.pop('cls', None) # type: ClsType["_models.FeatureOperationsListResult"] 134 error_map = { 135 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 136 } 137 error_map.update(kwargs.pop('error_map', {})) 138 api_version = "2021-07-01" 139 accept = "application/json, text/json" 140 141 def prepare_request(next_link=None): 142 # Construct headers 143 header_parameters = {} # type: Dict[str, Any] 144 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 145 146 if not next_link: 147 # Construct URL 148 url = self.list.metadata['url'] # type: ignore 149 path_format_arguments = { 150 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 151 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 152 } 153 url = self._client.format_url(url, **path_format_arguments) 154 # Construct parameters 155 query_parameters = {} # type: Dict[str, Any] 156 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 157 158 request = self._client.get(url, query_parameters, header_parameters) 159 else: 160 url = next_link 161 query_parameters = {} # type: Dict[str, Any] 162 request = self._client.get(url, query_parameters, header_parameters) 163 return request 164 165 def extract_data(pipeline_response): 166 deserialized = self._deserialize('FeatureOperationsListResult', pipeline_response) 167 list_of_elem = deserialized.value 168 if cls: 169 list_of_elem = cls(list_of_elem) 170 return deserialized.next_link or None, iter(list_of_elem) 171 172 def get_next(next_link=None): 173 request = prepare_request(next_link) 174 175 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 176 response = pipeline_response.http_response 177 178 if response.status_code not in [200]: 179 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 180 map_error(status_code=response.status_code, response=response, error_map=error_map) 181 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 182 183 return pipeline_response 184 185 return ItemPaged( 186 get_next, extract_data 187 ) 188 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Features/providers/{resourceProviderNamespace}/features'} # type: ignore 189 190 def get( 191 self, 192 resource_provider_namespace, # type: str 193 feature_name, # type: str 194 **kwargs # type: Any 195 ): 196 # type: (...) -> "_models.FeatureResult" 197 """Gets the preview feature with the specified name. 198 199 :param resource_provider_namespace: The resource provider namespace for the feature. 200 :type resource_provider_namespace: str 201 :param feature_name: The name of the feature to get. 202 :type feature_name: str 203 :keyword callable cls: A custom type or function that will be passed the direct response 204 :return: FeatureResult, or the result of cls(response) 205 :rtype: ~azure.mgmt.resource.features.v2021_07_01.models.FeatureResult 206 :raises: ~azure.core.exceptions.HttpResponseError 207 """ 208 cls = kwargs.pop('cls', None) # type: ClsType["_models.FeatureResult"] 209 error_map = { 210 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 211 } 212 error_map.update(kwargs.pop('error_map', {})) 213 api_version = "2021-07-01" 214 accept = "application/json, text/json" 215 216 # Construct URL 217 url = self.get.metadata['url'] # type: ignore 218 path_format_arguments = { 219 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 220 'featureName': self._serialize.url("feature_name", feature_name, 'str'), 221 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 222 } 223 url = self._client.format_url(url, **path_format_arguments) 224 225 # Construct parameters 226 query_parameters = {} # type: Dict[str, Any] 227 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 228 229 # Construct headers 230 header_parameters = {} # type: Dict[str, Any] 231 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 232 233 request = self._client.get(url, query_parameters, header_parameters) 234 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 235 response = pipeline_response.http_response 236 237 if response.status_code not in [200]: 238 map_error(status_code=response.status_code, response=response, error_map=error_map) 239 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 240 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 241 242 deserialized = self._deserialize('FeatureResult', pipeline_response) 243 244 if cls: 245 return cls(pipeline_response, deserialized, {}) 246 247 return deserialized 248 get.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Features/providers/{resourceProviderNamespace}/features/{featureName}'} # type: ignore 249 250 def register( 251 self, 252 resource_provider_namespace, # type: str 253 feature_name, # type: str 254 **kwargs # type: Any 255 ): 256 # type: (...) -> "_models.FeatureResult" 257 """Registers the preview feature for the subscription. 258 259 :param resource_provider_namespace: The namespace of the resource provider. 260 :type resource_provider_namespace: str 261 :param feature_name: The name of the feature to register. 262 :type feature_name: str 263 :keyword callable cls: A custom type or function that will be passed the direct response 264 :return: FeatureResult, or the result of cls(response) 265 :rtype: ~azure.mgmt.resource.features.v2021_07_01.models.FeatureResult 266 :raises: ~azure.core.exceptions.HttpResponseError 267 """ 268 cls = kwargs.pop('cls', None) # type: ClsType["_models.FeatureResult"] 269 error_map = { 270 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 271 } 272 error_map.update(kwargs.pop('error_map', {})) 273 api_version = "2021-07-01" 274 accept = "application/json, text/json" 275 276 # Construct URL 277 url = self.register.metadata['url'] # type: ignore 278 path_format_arguments = { 279 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 280 'featureName': self._serialize.url("feature_name", feature_name, 'str'), 281 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 282 } 283 url = self._client.format_url(url, **path_format_arguments) 284 285 # Construct parameters 286 query_parameters = {} # type: Dict[str, Any] 287 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 288 289 # Construct headers 290 header_parameters = {} # type: Dict[str, Any] 291 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 292 293 request = self._client.post(url, query_parameters, header_parameters) 294 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 295 response = pipeline_response.http_response 296 297 if response.status_code not in [200]: 298 map_error(status_code=response.status_code, response=response, error_map=error_map) 299 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 300 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 301 302 deserialized = self._deserialize('FeatureResult', pipeline_response) 303 304 if cls: 305 return cls(pipeline_response, deserialized, {}) 306 307 return deserialized 308 register.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Features/providers/{resourceProviderNamespace}/features/{featureName}/register'} # type: ignore 309 310 def unregister( 311 self, 312 resource_provider_namespace, # type: str 313 feature_name, # type: str 314 **kwargs # type: Any 315 ): 316 # type: (...) -> "_models.FeatureResult" 317 """Unregisters the preview feature for the subscription. 318 319 :param resource_provider_namespace: The namespace of the resource provider. 320 :type resource_provider_namespace: str 321 :param feature_name: The name of the feature to unregister. 322 :type feature_name: str 323 :keyword callable cls: A custom type or function that will be passed the direct response 324 :return: FeatureResult, or the result of cls(response) 325 :rtype: ~azure.mgmt.resource.features.v2021_07_01.models.FeatureResult 326 :raises: ~azure.core.exceptions.HttpResponseError 327 """ 328 cls = kwargs.pop('cls', None) # type: ClsType["_models.FeatureResult"] 329 error_map = { 330 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 331 } 332 error_map.update(kwargs.pop('error_map', {})) 333 api_version = "2021-07-01" 334 accept = "application/json, text/json" 335 336 # Construct URL 337 url = self.unregister.metadata['url'] # type: ignore 338 path_format_arguments = { 339 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 340 'featureName': self._serialize.url("feature_name", feature_name, 'str'), 341 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 342 } 343 url = self._client.format_url(url, **path_format_arguments) 344 345 # Construct parameters 346 query_parameters = {} # type: Dict[str, Any] 347 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 348 349 # Construct headers 350 header_parameters = {} # type: Dict[str, Any] 351 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 352 353 request = self._client.post(url, query_parameters, header_parameters) 354 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 355 response = pipeline_response.http_response 356 357 if response.status_code not in [200]: 358 map_error(status_code=response.status_code, response=response, error_map=error_map) 359 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 360 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 361 362 deserialized = self._deserialize('FeatureResult', pipeline_response) 363 364 if cls: 365 return cls(pipeline_response, deserialized, {}) 366 367 return deserialized 368 unregister.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Features/providers/{resourceProviderNamespace}/features/{featureName}/unregister'} # type: ignore 369