« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:45 +0000
1# -*- coding: utf-8 -*-
2# Copyright 2023 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#
17from google.auth.transport.requests import AuthorizedSession # type: ignore
18import json # type: ignore
19import grpc # type: ignore
20from google.auth.transport.grpc import SslCredentials # type: ignore
21from google.auth import credentials as ga_credentials # type: ignore
22from google.api_core import exceptions as core_exceptions
23from google.api_core import retry as retries
24from google.api_core import rest_helpers
25from google.api_core import rest_streaming
26from google.api_core import path_template
27from google.api_core import gapic_v1
29from google.protobuf import json_format
30from requests import __version__ as requests_version
31import dataclasses
32import re
33from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
34import warnings
36try:
37 OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault]
38except AttributeError: # pragma: NO COVER
39 OptionalRetry = Union[retries.Retry, object] # type: ignore
42from google.cloud.errorreporting_v1beta1.types import report_errors_service
44from .base import (
45 ReportErrorsServiceTransport,
46 DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO,
47)
50DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
51 gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
52 grpc_version=None,
53 rest_version=requests_version,
54)
57class ReportErrorsServiceRestInterceptor:
58 """Interceptor for ReportErrorsService.
60 Interceptors are used to manipulate requests, request metadata, and responses
61 in arbitrary ways.
62 Example use cases include:
63 * Logging
64 * Verifying requests according to service or custom semantics
65 * Stripping extraneous information from responses
67 These use cases and more can be enabled by injecting an
68 instance of a custom subclass when constructing the ReportErrorsServiceRestTransport.
70 .. code-block:: python
71 class MyCustomReportErrorsServiceInterceptor(ReportErrorsServiceRestInterceptor):
72 def pre_report_error_event(self, request, metadata):
73 logging.log(f"Received request: {request}")
74 return request, metadata
76 def post_report_error_event(self, response):
77 logging.log(f"Received response: {response}")
78 return response
80 transport = ReportErrorsServiceRestTransport(interceptor=MyCustomReportErrorsServiceInterceptor())
81 client = ReportErrorsServiceClient(transport=transport)
84 """
86 def pre_report_error_event(
87 self,
88 request: report_errors_service.ReportErrorEventRequest,
89 metadata: Sequence[Tuple[str, str]],
90 ) -> Tuple[
91 report_errors_service.ReportErrorEventRequest, Sequence[Tuple[str, str]]
92 ]:
93 """Pre-rpc interceptor for report_error_event
95 Override in a subclass to manipulate the request or metadata
96 before they are sent to the ReportErrorsService server.
97 """
98 return request, metadata
100 def post_report_error_event(
101 self, response: report_errors_service.ReportErrorEventResponse
102 ) -> report_errors_service.ReportErrorEventResponse:
103 """Post-rpc interceptor for report_error_event
105 Override in a subclass to manipulate the response
106 after it is returned by the ReportErrorsService server but before
107 it is returned to user code.
108 """
109 return response
112@dataclasses.dataclass
113class ReportErrorsServiceRestStub:
114 _session: AuthorizedSession
115 _host: str
116 _interceptor: ReportErrorsServiceRestInterceptor
119class ReportErrorsServiceRestTransport(ReportErrorsServiceTransport):
120 """REST backend transport for ReportErrorsService.
122 An API for reporting error events.
124 This class defines the same methods as the primary client, so the
125 primary client can load the underlying transport implementation
126 and call it.
128 It sends JSON representations of protocol buffers over HTTP/1.1
130 """
132 def __init__(
133 self,
134 *,
135 host: str = "clouderrorreporting.googleapis.com",
136 credentials: Optional[ga_credentials.Credentials] = None,
137 credentials_file: Optional[str] = None,
138 scopes: Optional[Sequence[str]] = None,
139 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
140 quota_project_id: Optional[str] = None,
141 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
142 always_use_jwt_access: Optional[bool] = False,
143 url_scheme: str = "https",
144 interceptor: Optional[ReportErrorsServiceRestInterceptor] = None,
145 api_audience: Optional[str] = None,
146 ) -> None:
147 """Instantiate the transport.
149 Args:
150 host (Optional[str]):
151 The hostname to connect to.
152 credentials (Optional[google.auth.credentials.Credentials]): The
153 authorization credentials to attach to requests. These
154 credentials identify the application to the service; if none
155 are specified, the client will attempt to ascertain the
156 credentials from the environment.
158 credentials_file (Optional[str]): A file with credentials that can
159 be loaded with :func:`google.auth.load_credentials_from_file`.
160 This argument is ignored if ``channel`` is provided.
161 scopes (Optional(Sequence[str])): A list of scopes. This argument is
162 ignored if ``channel`` is provided.
163 client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
164 certificate to configure mutual TLS HTTP channel. It is ignored
165 if ``channel`` is provided.
166 quota_project_id (Optional[str]): An optional project to use for billing
167 and quota.
168 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
169 The client info used to send a user-agent string along with
170 API requests. If ``None``, then default info will be used.
171 Generally, you only need to set this if you are developing
172 your own client library.
173 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
174 be used for service account credentials.
175 url_scheme: the protocol scheme for the API endpoint. Normally
176 "https", but for testing or local servers,
177 "http" can be specified.
178 """
179 # Run the base constructor
180 # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
181 # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
182 # credentials object
183 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
184 if maybe_url_match is None:
185 raise ValueError(
186 f"Unexpected hostname structure: {host}"
187 ) # pragma: NO COVER
189 url_match_items = maybe_url_match.groupdict()
191 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
193 super().__init__(
194 host=host,
195 credentials=credentials,
196 client_info=client_info,
197 always_use_jwt_access=always_use_jwt_access,
198 api_audience=api_audience,
199 )
200 self._session = AuthorizedSession(
201 self._credentials, default_host=self.DEFAULT_HOST
202 )
203 if client_cert_source_for_mtls:
204 self._session.configure_mtls_channel(client_cert_source_for_mtls)
205 self._interceptor = interceptor or ReportErrorsServiceRestInterceptor()
206 self._prep_wrapped_messages(client_info)
208 class _ReportErrorEvent(ReportErrorsServiceRestStub):
209 def __hash__(self):
210 return hash("ReportErrorEvent")
212 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
214 @classmethod
215 def _get_unset_required_fields(cls, message_dict):
216 return {
217 k: v
218 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
219 if k not in message_dict
220 }
222 def __call__(
223 self,
224 request: report_errors_service.ReportErrorEventRequest,
225 *,
226 retry: OptionalRetry = gapic_v1.method.DEFAULT,
227 timeout: Optional[float] = None,
228 metadata: Sequence[Tuple[str, str]] = (),
229 ) -> report_errors_service.ReportErrorEventResponse:
230 r"""Call the report error event method over HTTP.
232 Args:
233 request (~.report_errors_service.ReportErrorEventRequest):
234 The request object. A request for reporting an individual
235 error event.
236 retry (google.api_core.retry.Retry): Designation of what errors, if any,
237 should be retried.
238 timeout (float): The timeout for this request.
239 metadata (Sequence[Tuple[str, str]]): Strings which should be
240 sent along with the request as metadata.
242 Returns:
243 ~.report_errors_service.ReportErrorEventResponse:
244 Response for reporting an individual
245 error event. Data may be added to this
246 message in the future.
248 """
250 http_options: List[Dict[str, str]] = [
251 {
252 "method": "post",
253 "uri": "/v1beta1/{project_name=projects/*}/events:report",
254 "body": "event",
255 },
256 ]
257 request, metadata = self._interceptor.pre_report_error_event(
258 request, metadata
259 )
260 pb_request = report_errors_service.ReportErrorEventRequest.pb(request)
261 transcoded_request = path_template.transcode(http_options, pb_request)
263 # Jsonify the request body
265 body = json_format.MessageToJson(
266 transcoded_request["body"],
267 including_default_value_fields=False,
268 use_integers_for_enums=True,
269 )
270 uri = transcoded_request["uri"]
271 method = transcoded_request["method"]
273 # Jsonify the query params
274 query_params = json.loads(
275 json_format.MessageToJson(
276 transcoded_request["query_params"],
277 including_default_value_fields=False,
278 use_integers_for_enums=True,
279 )
280 )
281 query_params.update(self._get_unset_required_fields(query_params))
283 query_params["$alt"] = "json;enum-encoding=int"
285 # Send the request
286 headers = dict(metadata)
287 headers["Content-Type"] = "application/json"
288 response = getattr(self._session, method)(
289 "{host}{uri}".format(host=self._host, uri=uri),
290 timeout=timeout,
291 headers=headers,
292 params=rest_helpers.flatten_query_params(query_params, strict=True),
293 data=body,
294 )
296 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
297 # subclass.
298 if response.status_code >= 400:
299 raise core_exceptions.from_http_response(response)
301 # Return the response
302 resp = report_errors_service.ReportErrorEventResponse()
303 pb_resp = report_errors_service.ReportErrorEventResponse.pb(resp)
305 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
306 resp = self._interceptor.post_report_error_event(resp)
307 return resp
309 @property
310 def report_error_event(
311 self,
312 ) -> Callable[
313 [report_errors_service.ReportErrorEventRequest],
314 report_errors_service.ReportErrorEventResponse,
315 ]:
316 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
317 # In C++ this would require a dynamic_cast
318 return self._ReportErrorEvent(self._session, self._host, self._interceptor) # type: ignore
320 @property
321 def kind(self) -> str:
322 return "rest"
324 def close(self):
325 self._session.close()
328__all__ = ("ReportErrorsServiceRestTransport",)