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

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 ReportErrorsServiceTransport, 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 report_errors_service 

28 

29 

30class _BaseReportErrorsServiceRestTransport(ReportErrorsServiceTransport): 

31 """Base REST backend transport for ReportErrorsService. 

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 _BaseReportErrorEvent: 

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": "post", 

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

112 "body": "event", 

113 }, 

114 ] 

115 return http_options 

116 

117 @staticmethod 

118 def _get_transcoded_request(http_options, request): 

119 pb_request = report_errors_service.ReportErrorEventRequest.pb(request) 

120 transcoded_request = path_template.transcode(http_options, pb_request) 

121 return transcoded_request 

122 

123 @staticmethod 

124 def _get_request_body_json(transcoded_request): 

125 # Jsonify the request body 

126 

127 body = json_format.MessageToJson( 

128 transcoded_request["body"], use_integers_for_enums=True 

129 ) 

130 return body 

131 

132 @staticmethod 

133 def _get_query_params_json(transcoded_request): 

134 query_params = json.loads( 

135 json_format.MessageToJson( 

136 transcoded_request["query_params"], 

137 use_integers_for_enums=True, 

138 ) 

139 ) 

140 query_params.update( 

141 _BaseReportErrorsServiceRestTransport._BaseReportErrorEvent._get_unset_required_fields( 

142 query_params 

143 ) 

144 ) 

145 

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

147 return query_params 

148 

149 

150__all__ = ("_BaseReportErrorsServiceRestTransport",)