Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/errorreporting_v1beta1/services/error_group_service/transports/rest_base.py: 55%

1# -*- coding: utf-8 -*- 

2# Copyright 2025 Google LLC 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16import json # type: ignore 

17from google.api_core import path_template 

18from google.api_core import gapic_v1 

19 

20from google.protobuf import json_format 

21from .base import ErrorGroupServiceTransport, DEFAULT_CLIENT_INFO 

22 

23import re 

24from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union 

25 

26 

27from google.cloud.errorreporting_v1beta1.types import common 

28from google.cloud.errorreporting_v1beta1.types import error_group_service 

29 

30 

31class _BaseErrorGroupServiceRestTransport(ErrorGroupServiceTransport): 

32 """Base REST backend transport for ErrorGroupService. 

33 

34 Note: This class is not meant to be used directly. Use its sync and 

35 async sub-classes instead. 

36 

37 This class defines the same methods as the primary client, so the 

38 primary client can load the underlying transport implementation 

39 and call it. 

40 

41 It sends JSON representations of protocol buffers over HTTP/1.1 

42 """ 

43 

44 def __init__( 

45 self, 

46 *, 

47 host: str = "clouderrorreporting.googleapis.com", 

48 credentials: Optional[Any] = None, 

49 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

50 always_use_jwt_access: Optional[bool] = False, 

51 url_scheme: str = "https", 

52 api_audience: Optional[str] = None, 

53 ) -> None: 

54 """Instantiate the transport. 

55 Args: 

56 host (Optional[str]): 

57 The hostname to connect to (default: 'clouderrorreporting.googleapis.com'). 

58 credentials (Optional[Any]): The 

59 authorization credentials to attach to requests. These 

60 credentials identify the application to the service; if none 

61 are specified, the client will attempt to ascertain the 

62 credentials from the environment. 

63 client_info (google.api_core.gapic_v1.client_info.ClientInfo): 

64 The client info used to send a user-agent string along with 

65 API requests. If ``None``, then default info will be used. 

66 Generally, you only need to set this if you are developing 

67 your own client library. 

68 always_use_jwt_access (Optional[bool]): Whether self signed JWT should 

69 be used for service account credentials. 

70 url_scheme: the protocol scheme for the API endpoint. Normally 

71 "https", but for testing or local servers, 

72 "http" can be specified. 

73 """ 

74 # Run the base constructor 

75 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host) 

76 if maybe_url_match is None: 

77 raise ValueError( 

78 f"Unexpected hostname structure: {host}" 

79 ) # pragma: NO COVER 

80 

81 url_match_items = maybe_url_match.groupdict() 

82 

83 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host 

84 

85 super().__init__( 

86 host=host, 

87 credentials=credentials, 

88 client_info=client_info, 

89 always_use_jwt_access=always_use_jwt_access, 

90 api_audience=api_audience, 

91 ) 

92 

93 class _BaseGetGroup: 

94 def __hash__(self): # pragma: NO COVER 

95 return NotImplementedError("__hash__ must be implemented.") 

96 

97 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

98 

99 @classmethod 

100 def _get_unset_required_fields(cls, message_dict): 

101 return { 

102 k: v 

103 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

104 if k not in message_dict 

105 } 

106 

107 @staticmethod 

108 def _get_http_options(): 

109 http_options: List[Dict[str, str]] = [ 

110 { 

111 "method": "get", 

112 "uri": "/v1beta1/{group_name=projects/*/groups/*}", 

113 }, 

114 { 

115 "method": "get", 

116 "uri": "/v1beta1/{group_name=projects/*/locations/*/groups/*}", 

117 }, 

118 ] 

119 return http_options 

120 

121 @staticmethod 

122 def _get_transcoded_request(http_options, request): 

123 pb_request = error_group_service.GetGroupRequest.pb(request) 

124 transcoded_request = path_template.transcode(http_options, pb_request) 

125 return transcoded_request 

126 

127 @staticmethod 

128 def _get_query_params_json(transcoded_request): 

129 query_params = json.loads( 

130 json_format.MessageToJson( 

131 transcoded_request["query_params"], 

132 use_integers_for_enums=True, 

133 ) 

134 ) 

135 query_params.update( 

136 _BaseErrorGroupServiceRestTransport._BaseGetGroup._get_unset_required_fields( 

137 query_params 

138 ) 

139 ) 

140 

141 query_params["$alt"] = "json;enum-encoding=int" 

142 return query_params 

143 

144 class _BaseUpdateGroup: 

145 def __hash__(self): # pragma: NO COVER 

146 return NotImplementedError("__hash__ must be implemented.") 

147 

148 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

149 

150 @classmethod 

151 def _get_unset_required_fields(cls, message_dict): 

152 return { 

153 k: v 

154 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

155 if k not in message_dict 

156 } 

157 

158 @staticmethod 

159 def _get_http_options(): 

160 http_options: List[Dict[str, str]] = [ 

161 { 

162 "method": "put", 

163 "uri": "/v1beta1/{group.name=projects/*/groups/*}", 

164 "body": "group", 

165 }, 

166 { 

167 "method": "put", 

168 "uri": "/v1beta1/{group.name=projects/*/locations/*/groups/*}", 

169 "body": "group", 

170 }, 

171 ] 

172 return http_options 

173 

174 @staticmethod 

175 def _get_transcoded_request(http_options, request): 

176 pb_request = error_group_service.UpdateGroupRequest.pb(request) 

177 transcoded_request = path_template.transcode(http_options, pb_request) 

178 return transcoded_request 

179 

180 @staticmethod 

181 def _get_request_body_json(transcoded_request): 

182 # Jsonify the request body 

183 

184 body = json_format.MessageToJson( 

185 transcoded_request["body"], use_integers_for_enums=True 

186 ) 

187 return body 

188 

189 @staticmethod 

190 def _get_query_params_json(transcoded_request): 

191 query_params = json.loads( 

192 json_format.MessageToJson( 

193 transcoded_request["query_params"], 

194 use_integers_for_enums=True, 

195 ) 

196 ) 

197 query_params.update( 

198 _BaseErrorGroupServiceRestTransport._BaseUpdateGroup._get_unset_required_fields( 

199 query_params 

200 ) 

201 ) 

202 

203 query_params["$alt"] = "json;enum-encoding=int" 

204 return query_params 

205 

206 

207__all__ = ("_BaseErrorGroupServiceRestTransport",)