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

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 ErrorStatsServiceTransport, 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 error_stats_service 

28 

29 

30class _BaseErrorStatsServiceRestTransport(ErrorStatsServiceTransport): 

31 """Base REST backend transport for ErrorStatsService. 

32 

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

34 async sub-classes instead. 

35 

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

37 primary client can load the underlying transport implementation 

38 and call it. 

39 

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

41 """ 

42 

43 def __init__( 

44 self, 

45 *, 

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

47 credentials: Optional[Any] = None, 

48 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

49 always_use_jwt_access: Optional[bool] = False, 

50 url_scheme: str = "https", 

51 api_audience: Optional[str] = None, 

52 ) -> None: 

53 """Instantiate the transport. 

54 Args: 

55 host (Optional[str]): 

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

57 credentials (Optional[Any]): The 

58 authorization credentials to attach to requests. These 

59 credentials identify the application to the service; if none 

60 are specified, the client will attempt to ascertain the 

61 credentials from the environment. 

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

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

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

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

66 your own client library. 

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

68 be used for service account credentials. 

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

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

71 "http" can be specified. 

72 """ 

73 # Run the base constructor 

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

75 if maybe_url_match is None: 

76 raise ValueError( 

77 f"Unexpected hostname structure: {host}" 

78 ) # pragma: NO COVER 

79 

80 url_match_items = maybe_url_match.groupdict() 

81 

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

83 

84 super().__init__( 

85 host=host, 

86 credentials=credentials, 

87 client_info=client_info, 

88 always_use_jwt_access=always_use_jwt_access, 

89 api_audience=api_audience, 

90 ) 

91 

92 class _BaseDeleteEvents: 

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

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

95 

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

97 

98 @classmethod 

99 def _get_unset_required_fields(cls, message_dict): 

100 return { 

101 k: v 

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

103 if k not in message_dict 

104 } 

105 

106 @staticmethod 

107 def _get_http_options(): 

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

109 { 

110 "method": "delete", 

111 "uri": "/v1beta1/{project_name=projects/*}/events", 

112 }, 

113 { 

114 "method": "delete", 

115 "uri": "/v1beta1/{project_name=projects/*/locations/*}/events", 

116 }, 

117 ] 

118 return http_options 

119 

120 @staticmethod 

121 def _get_transcoded_request(http_options, request): 

122 pb_request = error_stats_service.DeleteEventsRequest.pb(request) 

123 transcoded_request = path_template.transcode(http_options, pb_request) 

124 return transcoded_request 

125 

126 @staticmethod 

127 def _get_query_params_json(transcoded_request): 

128 query_params = json.loads( 

129 json_format.MessageToJson( 

130 transcoded_request["query_params"], 

131 use_integers_for_enums=True, 

132 ) 

133 ) 

134 query_params.update( 

135 _BaseErrorStatsServiceRestTransport._BaseDeleteEvents._get_unset_required_fields( 

136 query_params 

137 ) 

138 ) 

139 

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

141 return query_params 

142 

143 class _BaseListEvents: 

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

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

146 

147 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = { 

148 "groupId": "", 

149 } 

150 

151 @classmethod 

152 def _get_unset_required_fields(cls, message_dict): 

153 return { 

154 k: v 

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

156 if k not in message_dict 

157 } 

158 

159 @staticmethod 

160 def _get_http_options(): 

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

162 { 

163 "method": "get", 

164 "uri": "/v1beta1/{project_name=projects/*}/events", 

165 }, 

166 { 

167 "method": "get", 

168 "uri": "/v1beta1/{project_name=projects/*/locations/*}/events", 

169 }, 

170 ] 

171 return http_options 

172 

173 @staticmethod 

174 def _get_transcoded_request(http_options, request): 

175 pb_request = error_stats_service.ListEventsRequest.pb(request) 

176 transcoded_request = path_template.transcode(http_options, pb_request) 

177 return transcoded_request 

178 

179 @staticmethod 

180 def _get_query_params_json(transcoded_request): 

181 query_params = json.loads( 

182 json_format.MessageToJson( 

183 transcoded_request["query_params"], 

184 use_integers_for_enums=True, 

185 ) 

186 ) 

187 query_params.update( 

188 _BaseErrorStatsServiceRestTransport._BaseListEvents._get_unset_required_fields( 

189 query_params 

190 ) 

191 ) 

192 

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

194 return query_params 

195 

196 class _BaseListGroupStats: 

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

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

199 

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

201 

202 @classmethod 

203 def _get_unset_required_fields(cls, message_dict): 

204 return { 

205 k: v 

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

207 if k not in message_dict 

208 } 

209 

210 @staticmethod 

211 def _get_http_options(): 

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

213 { 

214 "method": "get", 

215 "uri": "/v1beta1/{project_name=projects/*}/groupStats", 

216 }, 

217 { 

218 "method": "get", 

219 "uri": "/v1beta1/{project_name=projects/*/locations/*}/groupStats", 

220 }, 

221 ] 

222 return http_options 

223 

224 @staticmethod 

225 def _get_transcoded_request(http_options, request): 

226 pb_request = error_stats_service.ListGroupStatsRequest.pb(request) 

227 transcoded_request = path_template.transcode(http_options, pb_request) 

228 return transcoded_request 

229 

230 @staticmethod 

231 def _get_query_params_json(transcoded_request): 

232 query_params = json.loads( 

233 json_format.MessageToJson( 

234 transcoded_request["query_params"], 

235 use_integers_for_enums=True, 

236 ) 

237 ) 

238 query_params.update( 

239 _BaseErrorStatsServiceRestTransport._BaseListGroupStats._get_unset_required_fields( 

240 query_params 

241 ) 

242 ) 

243 

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

245 return query_params 

246 

247 

248__all__ = ("_BaseErrorStatsServiceRestTransport",)