Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# pylint: disable=too-many-lines
2# coding=utf-8
3# --------------------------------------------------------------------------
4# Copyright (c) Microsoft Corporation. All rights reserved.
5# Licensed under the MIT License. See License.txt in the project root for license information.
6# Code generated by Microsoft (R) AutoRest Code Generator.
7# Changes may cause incorrect behavior and will be lost if the code is regenerated.
8# --------------------------------------------------------------------------
9from io import IOBase
10from typing import Any, Callable, Dict, IO, Iterable, Optional, TypeVar, Union, cast, overload
11import urllib.parse
13from azure.core.exceptions import (
14 ClientAuthenticationError,
15 HttpResponseError,
16 ResourceExistsError,
17 ResourceNotFoundError,
18 ResourceNotModifiedError,
19 map_error,
20)
21from azure.core.paging import ItemPaged
22from azure.core.pipeline import PipelineResponse
23from azure.core.pipeline.transport import HttpResponse
24from azure.core.polling import LROPoller, NoPolling, PollingMethod
25from azure.core.rest import HttpRequest
26from azure.core.tracing.decorator import distributed_trace
27from azure.core.utils import case_insensitive_dict
28from azure.mgmt.core.exceptions import ARMErrorFormat
29from azure.mgmt.core.polling.arm_polling import ARMPolling
31from .. import models as _models
32from .._serialization import Serializer
33from .._vendor import _convert_request
35T = TypeVar("T")
36ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
38_SERIALIZER = Serializer()
39_SERIALIZER.client_side_validation = False
42def build_create_or_update_request(
43 resource_group_name: str, monitor_name: str, configuration_name: str, subscription_id: str, **kwargs: Any
44) -> HttpRequest:
45 _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
46 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
48 api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2023-04-27"))
49 content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
50 accept = _headers.pop("Accept", "application/json")
52 # Construct URL
53 _url = kwargs.pop(
54 "template_url",
55 "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations/{configurationName}",
56 ) # pylint: disable=line-too-long
57 path_format_arguments = {
58 "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str", min_length=1),
59 "resourceGroupName": _SERIALIZER.url(
60 "resource_group_name", resource_group_name, "str", max_length=90, min_length=1
61 ),
62 "monitorName": _SERIALIZER.url("monitor_name", monitor_name, "str"),
63 "configurationName": _SERIALIZER.url("configuration_name", configuration_name, "str"),
64 }
66 _url: str = _url.format(**path_format_arguments) # type: ignore
68 # Construct parameters
69 _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
71 # Construct headers
72 if content_type is not None:
73 _headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
74 _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
76 return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
79def build_get_request(
80 resource_group_name: str, monitor_name: str, configuration_name: str, subscription_id: str, **kwargs: Any
81) -> HttpRequest:
82 _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
83 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
85 api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2023-04-27"))
86 accept = _headers.pop("Accept", "application/json")
88 # Construct URL
89 _url = kwargs.pop(
90 "template_url",
91 "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations/{configurationName}",
92 ) # pylint: disable=line-too-long
93 path_format_arguments = {
94 "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str", min_length=1),
95 "resourceGroupName": _SERIALIZER.url(
96 "resource_group_name", resource_group_name, "str", max_length=90, min_length=1
97 ),
98 "monitorName": _SERIALIZER.url("monitor_name", monitor_name, "str"),
99 "configurationName": _SERIALIZER.url("configuration_name", configuration_name, "str"),
100 }
102 _url: str = _url.format(**path_format_arguments) # type: ignore
104 # Construct parameters
105 _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
107 # Construct headers
108 _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
110 return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
113def build_list_request(resource_group_name: str, monitor_name: str, subscription_id: str, **kwargs: Any) -> HttpRequest:
114 _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
115 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
117 api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2023-04-27"))
118 accept = _headers.pop("Accept", "application/json")
120 # Construct URL
121 _url = kwargs.pop(
122 "template_url",
123 "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations",
124 ) # pylint: disable=line-too-long
125 path_format_arguments = {
126 "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str", min_length=1),
127 "resourceGroupName": _SERIALIZER.url(
128 "resource_group_name", resource_group_name, "str", max_length=90, min_length=1
129 ),
130 "monitorName": _SERIALIZER.url("monitor_name", monitor_name, "str"),
131 }
133 _url: str = _url.format(**path_format_arguments) # type: ignore
135 # Construct parameters
136 _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
138 # Construct headers
139 _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
141 return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
144class SingleSignOnOperations:
145 """
146 .. warning::
147 **DO NOT** instantiate this class directly.
149 Instead, you should access the following operations through
150 :class:`~azure.mgmt.dynatrace.DynatraceObservabilityMgmtClient`'s
151 :attr:`single_sign_on` attribute.
152 """
154 models = _models
156 def __init__(self, *args, **kwargs):
157 input_args = list(args)
158 self._client = input_args.pop(0) if input_args else kwargs.pop("client")
159 self._config = input_args.pop(0) if input_args else kwargs.pop("config")
160 self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer")
161 self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer")
163 def _create_or_update_initial(
164 self,
165 resource_group_name: str,
166 monitor_name: str,
167 configuration_name: str,
168 resource: Union[_models.DynatraceSingleSignOnResource, IO],
169 **kwargs: Any
170 ) -> _models.DynatraceSingleSignOnResource:
171 error_map = {
172 401: ClientAuthenticationError,
173 404: ResourceNotFoundError,
174 409: ResourceExistsError,
175 304: ResourceNotModifiedError,
176 }
177 error_map.update(kwargs.pop("error_map", {}) or {})
179 _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
180 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
182 api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
183 content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
184 cls: ClsType[_models.DynatraceSingleSignOnResource] = kwargs.pop("cls", None)
186 content_type = content_type or "application/json"
187 _json = None
188 _content = None
189 if isinstance(resource, (IOBase, bytes)):
190 _content = resource
191 else:
192 _json = self._serialize.body(resource, "DynatraceSingleSignOnResource")
194 request = build_create_or_update_request(
195 resource_group_name=resource_group_name,
196 monitor_name=monitor_name,
197 configuration_name=configuration_name,
198 subscription_id=self._config.subscription_id,
199 api_version=api_version,
200 content_type=content_type,
201 json=_json,
202 content=_content,
203 template_url=self._create_or_update_initial.metadata["url"],
204 headers=_headers,
205 params=_params,
206 )
207 request = _convert_request(request)
208 request.url = self._client.format_url(request.url)
210 _stream = False
211 pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
212 request, stream=_stream, **kwargs
213 )
215 response = pipeline_response.http_response
217 if response.status_code not in [200, 201]:
218 map_error(status_code=response.status_code, response=response, error_map=error_map)
219 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
220 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
222 if response.status_code == 200:
223 deserialized = self._deserialize("DynatraceSingleSignOnResource", pipeline_response)
225 if response.status_code == 201:
226 deserialized = self._deserialize("DynatraceSingleSignOnResource", pipeline_response)
228 if cls:
229 return cls(pipeline_response, deserialized, {}) # type: ignore
231 return deserialized # type: ignore
233 _create_or_update_initial.metadata = {
234 "url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations/{configurationName}"
235 }
237 @overload
238 def begin_create_or_update(
239 self,
240 resource_group_name: str,
241 monitor_name: str,
242 configuration_name: str,
243 resource: _models.DynatraceSingleSignOnResource,
244 *,
245 content_type: str = "application/json",
246 **kwargs: Any
247 ) -> LROPoller[_models.DynatraceSingleSignOnResource]:
248 """Create a DynatraceSingleSignOnResource.
250 Create a DynatraceSingleSignOnResource.
252 :param resource_group_name: The name of the resource group. The name is case insensitive.
253 Required.
254 :type resource_group_name: str
255 :param monitor_name: Monitor resource name. Required.
256 :type monitor_name: str
257 :param configuration_name: Single Sign On Configuration Name. Required.
258 :type configuration_name: str
259 :param resource: Resource create parameters. Required.
260 :type resource: ~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource
261 :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
262 Default value is "application/json".
263 :paramtype content_type: str
264 :keyword callable cls: A custom type or function that will be passed the direct response
265 :keyword str continuation_token: A continuation token to restart a poller from a saved state.
266 :keyword polling: By default, your polling method will be ARMPolling. Pass in False for this
267 operation to not poll, or pass in your own initialized polling object for a personal polling
268 strategy.
269 :paramtype polling: bool or ~azure.core.polling.PollingMethod
270 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no
271 Retry-After header is present.
272 :return: An instance of LROPoller that returns either DynatraceSingleSignOnResource or the
273 result of cls(response)
274 :rtype:
275 ~azure.core.polling.LROPoller[~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource]
276 :raises ~azure.core.exceptions.HttpResponseError:
277 """
279 @overload
280 def begin_create_or_update(
281 self,
282 resource_group_name: str,
283 monitor_name: str,
284 configuration_name: str,
285 resource: IO,
286 *,
287 content_type: str = "application/json",
288 **kwargs: Any
289 ) -> LROPoller[_models.DynatraceSingleSignOnResource]:
290 """Create a DynatraceSingleSignOnResource.
292 Create a DynatraceSingleSignOnResource.
294 :param resource_group_name: The name of the resource group. The name is case insensitive.
295 Required.
296 :type resource_group_name: str
297 :param monitor_name: Monitor resource name. Required.
298 :type monitor_name: str
299 :param configuration_name: Single Sign On Configuration Name. Required.
300 :type configuration_name: str
301 :param resource: Resource create parameters. Required.
302 :type resource: IO
303 :keyword content_type: Body Parameter content-type. Content type parameter for binary body.
304 Default value is "application/json".
305 :paramtype content_type: str
306 :keyword callable cls: A custom type or function that will be passed the direct response
307 :keyword str continuation_token: A continuation token to restart a poller from a saved state.
308 :keyword polling: By default, your polling method will be ARMPolling. Pass in False for this
309 operation to not poll, or pass in your own initialized polling object for a personal polling
310 strategy.
311 :paramtype polling: bool or ~azure.core.polling.PollingMethod
312 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no
313 Retry-After header is present.
314 :return: An instance of LROPoller that returns either DynatraceSingleSignOnResource or the
315 result of cls(response)
316 :rtype:
317 ~azure.core.polling.LROPoller[~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource]
318 :raises ~azure.core.exceptions.HttpResponseError:
319 """
321 @distributed_trace
322 def begin_create_or_update(
323 self,
324 resource_group_name: str,
325 monitor_name: str,
326 configuration_name: str,
327 resource: Union[_models.DynatraceSingleSignOnResource, IO],
328 **kwargs: Any
329 ) -> LROPoller[_models.DynatraceSingleSignOnResource]:
330 """Create a DynatraceSingleSignOnResource.
332 Create a DynatraceSingleSignOnResource.
334 :param resource_group_name: The name of the resource group. The name is case insensitive.
335 Required.
336 :type resource_group_name: str
337 :param monitor_name: Monitor resource name. Required.
338 :type monitor_name: str
339 :param configuration_name: Single Sign On Configuration Name. Required.
340 :type configuration_name: str
341 :param resource: Resource create parameters. Is either a DynatraceSingleSignOnResource type or
342 a IO type. Required.
343 :type resource: ~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource or IO
344 :keyword content_type: Body Parameter content-type. Known values are: 'application/json'.
345 Default value is None.
346 :paramtype content_type: str
347 :keyword callable cls: A custom type or function that will be passed the direct response
348 :keyword str continuation_token: A continuation token to restart a poller from a saved state.
349 :keyword polling: By default, your polling method will be ARMPolling. Pass in False for this
350 operation to not poll, or pass in your own initialized polling object for a personal polling
351 strategy.
352 :paramtype polling: bool or ~azure.core.polling.PollingMethod
353 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no
354 Retry-After header is present.
355 :return: An instance of LROPoller that returns either DynatraceSingleSignOnResource or the
356 result of cls(response)
357 :rtype:
358 ~azure.core.polling.LROPoller[~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource]
359 :raises ~azure.core.exceptions.HttpResponseError:
360 """
361 _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
362 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
364 api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
365 content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
366 cls: ClsType[_models.DynatraceSingleSignOnResource] = kwargs.pop("cls", None)
367 polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
368 lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
369 cont_token: Optional[str] = kwargs.pop("continuation_token", None)
370 if cont_token is None:
371 raw_result = self._create_or_update_initial(
372 resource_group_name=resource_group_name,
373 monitor_name=monitor_name,
374 configuration_name=configuration_name,
375 resource=resource,
376 api_version=api_version,
377 content_type=content_type,
378 cls=lambda x, y, z: x,
379 headers=_headers,
380 params=_params,
381 **kwargs
382 )
383 kwargs.pop("error_map", None)
385 def get_long_running_output(pipeline_response):
386 deserialized = self._deserialize("DynatraceSingleSignOnResource", pipeline_response)
387 if cls:
388 return cls(pipeline_response, deserialized, {})
389 return deserialized
391 if polling is True:
392 polling_method: PollingMethod = cast(
393 PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "azure-async-operation"}, **kwargs)
394 )
395 elif polling is False:
396 polling_method = cast(PollingMethod, NoPolling())
397 else:
398 polling_method = polling
399 if cont_token:
400 return LROPoller.from_continuation_token(
401 polling_method=polling_method,
402 continuation_token=cont_token,
403 client=self._client,
404 deserialization_callback=get_long_running_output,
405 )
406 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) # type: ignore
408 begin_create_or_update.metadata = {
409 "url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations/{configurationName}"
410 }
412 @distributed_trace
413 def get(
414 self, resource_group_name: str, monitor_name: str, configuration_name: str, **kwargs: Any
415 ) -> _models.DynatraceSingleSignOnResource:
416 """Get a DynatraceSingleSignOnResource.
418 Get a DynatraceSingleSignOnResource.
420 :param resource_group_name: The name of the resource group. The name is case insensitive.
421 Required.
422 :type resource_group_name: str
423 :param monitor_name: Monitor resource name. Required.
424 :type monitor_name: str
425 :param configuration_name: Single Sign On Configuration Name. Required.
426 :type configuration_name: str
427 :keyword callable cls: A custom type or function that will be passed the direct response
428 :return: DynatraceSingleSignOnResource or the result of cls(response)
429 :rtype: ~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource
430 :raises ~azure.core.exceptions.HttpResponseError:
431 """
432 error_map = {
433 401: ClientAuthenticationError,
434 404: ResourceNotFoundError,
435 409: ResourceExistsError,
436 304: ResourceNotModifiedError,
437 }
438 error_map.update(kwargs.pop("error_map", {}) or {})
440 _headers = kwargs.pop("headers", {}) or {}
441 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
443 api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
444 cls: ClsType[_models.DynatraceSingleSignOnResource] = kwargs.pop("cls", None)
446 request = build_get_request(
447 resource_group_name=resource_group_name,
448 monitor_name=monitor_name,
449 configuration_name=configuration_name,
450 subscription_id=self._config.subscription_id,
451 api_version=api_version,
452 template_url=self.get.metadata["url"],
453 headers=_headers,
454 params=_params,
455 )
456 request = _convert_request(request)
457 request.url = self._client.format_url(request.url)
459 _stream = False
460 pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
461 request, stream=_stream, **kwargs
462 )
464 response = pipeline_response.http_response
466 if response.status_code not in [200]:
467 map_error(status_code=response.status_code, response=response, error_map=error_map)
468 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
469 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
471 deserialized = self._deserialize("DynatraceSingleSignOnResource", pipeline_response)
473 if cls:
474 return cls(pipeline_response, deserialized, {})
476 return deserialized
478 get.metadata = {
479 "url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations/{configurationName}"
480 }
482 @distributed_trace
483 def list(
484 self, resource_group_name: str, monitor_name: str, **kwargs: Any
485 ) -> Iterable["_models.DynatraceSingleSignOnResource"]:
486 """List all DynatraceSingleSignOnResource by monitorName.
488 List all DynatraceSingleSignOnResource by monitorName.
490 :param resource_group_name: The name of the resource group. The name is case insensitive.
491 Required.
492 :type resource_group_name: str
493 :param monitor_name: Monitor resource name. Required.
494 :type monitor_name: str
495 :keyword callable cls: A custom type or function that will be passed the direct response
496 :return: An iterator like instance of either DynatraceSingleSignOnResource or the result of
497 cls(response)
498 :rtype:
499 ~azure.core.paging.ItemPaged[~azure.mgmt.dynatrace.models.DynatraceSingleSignOnResource]
500 :raises ~azure.core.exceptions.HttpResponseError:
501 """
502 _headers = kwargs.pop("headers", {}) or {}
503 _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
505 api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
506 cls: ClsType[_models.DynatraceSingleSignOnResourceListResult] = kwargs.pop("cls", None)
508 error_map = {
509 401: ClientAuthenticationError,
510 404: ResourceNotFoundError,
511 409: ResourceExistsError,
512 304: ResourceNotModifiedError,
513 }
514 error_map.update(kwargs.pop("error_map", {}) or {})
516 def prepare_request(next_link=None):
517 if not next_link:
519 request = build_list_request(
520 resource_group_name=resource_group_name,
521 monitor_name=monitor_name,
522 subscription_id=self._config.subscription_id,
523 api_version=api_version,
524 template_url=self.list.metadata["url"],
525 headers=_headers,
526 params=_params,
527 )
528 request = _convert_request(request)
529 request.url = self._client.format_url(request.url)
531 else:
532 # make call to next link with the client's api-version
533 _parsed_next_link = urllib.parse.urlparse(next_link)
534 _next_request_params = case_insensitive_dict(
535 {
536 key: [urllib.parse.quote(v) for v in value]
537 for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
538 }
539 )
540 _next_request_params["api-version"] = self._config.api_version
541 request = HttpRequest(
542 "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
543 )
544 request = _convert_request(request)
545 request.url = self._client.format_url(request.url)
546 request.method = "GET"
547 return request
549 def extract_data(pipeline_response):
550 deserialized = self._deserialize("DynatraceSingleSignOnResourceListResult", pipeline_response)
551 list_of_elem = deserialized.value
552 if cls:
553 list_of_elem = cls(list_of_elem) # type: ignore
554 return deserialized.next_link or None, iter(list_of_elem)
556 def get_next(next_link=None):
557 request = prepare_request(next_link)
559 _stream = False
560 pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
561 request, stream=_stream, **kwargs
562 )
563 response = pipeline_response.http_response
565 if response.status_code not in [200]:
566 map_error(status_code=response.status_code, response=response, error_map=error_map)
567 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, pipeline_response)
568 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
570 return pipeline_response
572 return ItemPaged(get_next, extract_data)
574 list.metadata = {
575 "url": "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Dynatrace.Observability/monitors/{monitorName}/singleSignOnConfigurations"
576 }