Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/azure/mgmt/dynatrace/_client.py: 40%

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

53 statements  

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) Python Code Generator. 

6# Changes may cause incorrect behavior and will be lost if the code is regenerated. 

7# -------------------------------------------------------------------------- 

8 

9from copy import deepcopy 

10import sys 

11from typing import Any, Optional, TYPE_CHECKING, cast 

12 

13from azure.core.pipeline import policies 

14from azure.core.rest import HttpRequest, HttpResponse 

15from azure.core.settings import settings 

16from azure.mgmt.core import ARMPipelineClient 

17from azure.mgmt.core.policies import ARMAutoResourceProviderRegistrationPolicy 

18from azure.mgmt.core.tools import get_arm_endpoints 

19 

20from ._configuration import DynatraceObservabilityMgmtClientConfiguration 

21from ._utils.serialization import Deserializer, Serializer 

22from .operations import ( 

23 CreationSupportedOperations, 

24 MonitoredSubscriptionsOperations, 

25 MonitorsOperations, 

26 Operations, 

27 SingleSignOnOperations, 

28 TagRulesOperations, 

29) 

30 

31if sys.version_info >= (3, 11): 

32 from typing import Self 

33else: 

34 from typing_extensions import Self # type: ignore 

35 

36if TYPE_CHECKING: 

37 from azure.core import AzureClouds 

38 from azure.core.credentials import TokenCredential 

39 

40 

41class DynatraceObservabilityMgmtClient: 

42 """DynatraceObservabilityMgmtClient. 

43 

44 :ivar operations: Operations operations 

45 :vartype operations: azure.mgmt.dynatrace.operations.Operations 

46 :ivar monitored_subscriptions: MonitoredSubscriptionsOperations operations 

47 :vartype monitored_subscriptions: 

48 azure.mgmt.dynatrace.operations.MonitoredSubscriptionsOperations 

49 :ivar tag_rules: TagRulesOperations operations 

50 :vartype tag_rules: azure.mgmt.dynatrace.operations.TagRulesOperations 

51 :ivar monitors: MonitorsOperations operations 

52 :vartype monitors: azure.mgmt.dynatrace.operations.MonitorsOperations 

53 :ivar single_sign_on: SingleSignOnOperations operations 

54 :vartype single_sign_on: azure.mgmt.dynatrace.operations.SingleSignOnOperations 

55 :ivar creation_supported: CreationSupportedOperations operations 

56 :vartype creation_supported: azure.mgmt.dynatrace.operations.CreationSupportedOperations 

57 :param credential: Credential used to authenticate requests to the service. Required. 

58 :type credential: ~azure.core.credentials.TokenCredential 

59 :param subscription_id: The ID of the target subscription. The value must be an UUID. Required. 

60 :type subscription_id: str 

61 :param base_url: Service host. Default value is None. 

62 :type base_url: str 

63 :keyword cloud_setting: The cloud setting for which to get the ARM endpoint. Default value is 

64 None. 

65 :paramtype cloud_setting: ~azure.core.AzureClouds 

66 :keyword api_version: The API version to use for this operation. Known values are "2024-04-24" 

67 and None. Default value is None. If not set, the operation's default API version will be used. 

68 Note that overriding this default value may result in unsupported behavior. 

69 :paramtype api_version: str 

70 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no 

71 Retry-After header is present. 

72 """ 

73 

74 def __init__( 

75 self, 

76 credential: "TokenCredential", 

77 subscription_id: str, 

78 base_url: Optional[str] = None, 

79 *, 

80 cloud_setting: Optional["AzureClouds"] = None, 

81 **kwargs: Any 

82 ) -> None: 

83 _endpoint = "{endpoint}" 

84 _cloud = cloud_setting or settings.current.azure_cloud # type: ignore 

85 _endpoints = get_arm_endpoints(_cloud) 

86 if not base_url: 

87 base_url = _endpoints["resource_manager"] 

88 credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"]) 

89 self._config = DynatraceObservabilityMgmtClientConfiguration( 

90 credential=credential, 

91 subscription_id=subscription_id, 

92 base_url=cast(str, base_url), 

93 cloud_setting=cloud_setting, 

94 credential_scopes=credential_scopes, 

95 **kwargs 

96 ) 

97 

98 _policies = kwargs.pop("policies", None) 

99 if _policies is None: 

100 _policies = [ 

101 policies.RequestIdPolicy(**kwargs), 

102 self._config.headers_policy, 

103 self._config.user_agent_policy, 

104 self._config.proxy_policy, 

105 policies.ContentDecodePolicy(**kwargs), 

106 ARMAutoResourceProviderRegistrationPolicy(), 

107 self._config.redirect_policy, 

108 self._config.retry_policy, 

109 self._config.authentication_policy, 

110 self._config.custom_hook_policy, 

111 self._config.logging_policy, 

112 policies.DistributedTracingPolicy(**kwargs), 

113 policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None, 

114 self._config.http_logging_policy, 

115 ] 

116 self._client: ARMPipelineClient = ARMPipelineClient(base_url=cast(str, _endpoint), policies=_policies, **kwargs) 

117 

118 self._serialize = Serializer() 

119 self._deserialize = Deserializer() 

120 self._serialize.client_side_validation = False 

121 self.operations = Operations(self._client, self._config, self._serialize, self._deserialize) 

122 self.monitored_subscriptions = MonitoredSubscriptionsOperations( 

123 self._client, self._config, self._serialize, self._deserialize 

124 ) 

125 self.tag_rules = TagRulesOperations(self._client, self._config, self._serialize, self._deserialize) 

126 self.monitors = MonitorsOperations(self._client, self._config, self._serialize, self._deserialize) 

127 self.single_sign_on = SingleSignOnOperations(self._client, self._config, self._serialize, self._deserialize) 

128 self.creation_supported = CreationSupportedOperations( 

129 self._client, self._config, self._serialize, self._deserialize 

130 ) 

131 

132 def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs: Any) -> HttpResponse: 

133 """Runs the network request through the client's chained policies. 

134 

135 >>> from azure.core.rest import HttpRequest 

136 >>> request = HttpRequest("GET", "https://www.example.org/") 

137 <HttpRequest [GET], url: 'https://www.example.org/'> 

138 >>> response = client.send_request(request) 

139 <HttpResponse: 200 OK> 

140 

141 For more information on this code flow, see https://aka.ms/azsdk/dpcodegen/python/send_request 

142 

143 :param request: The network request you want to make. Required. 

144 :type request: ~azure.core.rest.HttpRequest 

145 :keyword bool stream: Whether the response payload will be streamed. Defaults to False. 

146 :return: The response of your network call. Does not do error handling on your response. 

147 :rtype: ~azure.core.rest.HttpResponse 

148 """ 

149 

150 request_copy = deepcopy(request) 

151 path_format_arguments = { 

152 "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), 

153 } 

154 

155 request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments) 

156 return self._client.send_request(request_copy, stream=stream, **kwargs) # type: ignore 

157 

158 def close(self) -> None: 

159 self._client.close() 

160 

161 def __enter__(self) -> Self: 

162 self._client.__enter__() 

163 return self 

164 

165 def __exit__(self, *exc_details: Any) -> None: 

166 self._client.__exit__(*exc_details)