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
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 logging
17import json # type: ignore
19from google.auth.transport.requests import AuthorizedSession # type: ignore
20from google.auth import credentials as ga_credentials # type: ignore
21from google.api_core import exceptions as core_exceptions
22from google.api_core import retry as retries
23from google.api_core import rest_helpers
24from google.api_core import rest_streaming
25from google.api_core import gapic_v1
26import google.protobuf
28from google.protobuf import json_format
30from requests import __version__ as requests_version
31import dataclasses
32from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
33import warnings
36from google.cloud.errorreporting_v1beta1.types import error_stats_service
39from .rest_base import _BaseErrorStatsServiceRestTransport
40from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO
42try:
43 OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault, None]
44except AttributeError: # pragma: NO COVER
45 OptionalRetry = Union[retries.Retry, object, None] # type: ignore
47try:
48 from google.api_core import client_logging # type: ignore
50 CLIENT_LOGGING_SUPPORTED = True # pragma: NO COVER
51except ImportError: # pragma: NO COVER
52 CLIENT_LOGGING_SUPPORTED = False
54_LOGGER = logging.getLogger(__name__)
56DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
57 gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
58 grpc_version=None,
59 rest_version=f"requests@{requests_version}",
60)
62if hasattr(DEFAULT_CLIENT_INFO, "protobuf_runtime_version"): # pragma: NO COVER
63 DEFAULT_CLIENT_INFO.protobuf_runtime_version = google.protobuf.__version__
66class ErrorStatsServiceRestInterceptor:
67 """Interceptor for ErrorStatsService.
69 Interceptors are used to manipulate requests, request metadata, and responses
70 in arbitrary ways.
71 Example use cases include:
72 * Logging
73 * Verifying requests according to service or custom semantics
74 * Stripping extraneous information from responses
76 These use cases and more can be enabled by injecting an
77 instance of a custom subclass when constructing the ErrorStatsServiceRestTransport.
79 .. code-block:: python
80 class MyCustomErrorStatsServiceInterceptor(ErrorStatsServiceRestInterceptor):
81 def pre_delete_events(self, request, metadata):
82 logging.log(f"Received request: {request}")
83 return request, metadata
85 def post_delete_events(self, response):
86 logging.log(f"Received response: {response}")
87 return response
89 def pre_list_events(self, request, metadata):
90 logging.log(f"Received request: {request}")
91 return request, metadata
93 def post_list_events(self, response):
94 logging.log(f"Received response: {response}")
95 return response
97 def pre_list_group_stats(self, request, metadata):
98 logging.log(f"Received request: {request}")
99 return request, metadata
101 def post_list_group_stats(self, response):
102 logging.log(f"Received response: {response}")
103 return response
105 transport = ErrorStatsServiceRestTransport(interceptor=MyCustomErrorStatsServiceInterceptor())
106 client = ErrorStatsServiceClient(transport=transport)
109 """
111 def pre_delete_events(
112 self,
113 request: error_stats_service.DeleteEventsRequest,
114 metadata: Sequence[Tuple[str, Union[str, bytes]]],
115 ) -> Tuple[
116 error_stats_service.DeleteEventsRequest, Sequence[Tuple[str, Union[str, bytes]]]
117 ]:
118 """Pre-rpc interceptor for delete_events
120 Override in a subclass to manipulate the request or metadata
121 before they are sent to the ErrorStatsService server.
122 """
123 return request, metadata
125 def post_delete_events(
126 self, response: error_stats_service.DeleteEventsResponse
127 ) -> error_stats_service.DeleteEventsResponse:
128 """Post-rpc interceptor for delete_events
130 DEPRECATED. Please use the `post_delete_events_with_metadata`
131 interceptor instead.
133 Override in a subclass to read or manipulate the response
134 after it is returned by the ErrorStatsService server but before
135 it is returned to user code. This `post_delete_events` interceptor runs
136 before the `post_delete_events_with_metadata` interceptor.
137 """
138 return response
140 def post_delete_events_with_metadata(
141 self,
142 response: error_stats_service.DeleteEventsResponse,
143 metadata: Sequence[Tuple[str, Union[str, bytes]]],
144 ) -> Tuple[
145 error_stats_service.DeleteEventsResponse,
146 Sequence[Tuple[str, Union[str, bytes]]],
147 ]:
148 """Post-rpc interceptor for delete_events
150 Override in a subclass to read or manipulate the response or metadata after it
151 is returned by the ErrorStatsService server but before it is returned to user code.
153 We recommend only using this `post_delete_events_with_metadata`
154 interceptor in new development instead of the `post_delete_events` interceptor.
155 When both interceptors are used, this `post_delete_events_with_metadata` interceptor runs after the
156 `post_delete_events` interceptor. The (possibly modified) response returned by
157 `post_delete_events` will be passed to
158 `post_delete_events_with_metadata`.
159 """
160 return response, metadata
162 def pre_list_events(
163 self,
164 request: error_stats_service.ListEventsRequest,
165 metadata: Sequence[Tuple[str, Union[str, bytes]]],
166 ) -> Tuple[
167 error_stats_service.ListEventsRequest, Sequence[Tuple[str, Union[str, bytes]]]
168 ]:
169 """Pre-rpc interceptor for list_events
171 Override in a subclass to manipulate the request or metadata
172 before they are sent to the ErrorStatsService server.
173 """
174 return request, metadata
176 def post_list_events(
177 self, response: error_stats_service.ListEventsResponse
178 ) -> error_stats_service.ListEventsResponse:
179 """Post-rpc interceptor for list_events
181 DEPRECATED. Please use the `post_list_events_with_metadata`
182 interceptor instead.
184 Override in a subclass to read or manipulate the response
185 after it is returned by the ErrorStatsService server but before
186 it is returned to user code. This `post_list_events` interceptor runs
187 before the `post_list_events_with_metadata` interceptor.
188 """
189 return response
191 def post_list_events_with_metadata(
192 self,
193 response: error_stats_service.ListEventsResponse,
194 metadata: Sequence[Tuple[str, Union[str, bytes]]],
195 ) -> Tuple[
196 error_stats_service.ListEventsResponse, Sequence[Tuple[str, Union[str, bytes]]]
197 ]:
198 """Post-rpc interceptor for list_events
200 Override in a subclass to read or manipulate the response or metadata after it
201 is returned by the ErrorStatsService server but before it is returned to user code.
203 We recommend only using this `post_list_events_with_metadata`
204 interceptor in new development instead of the `post_list_events` interceptor.
205 When both interceptors are used, this `post_list_events_with_metadata` interceptor runs after the
206 `post_list_events` interceptor. The (possibly modified) response returned by
207 `post_list_events` will be passed to
208 `post_list_events_with_metadata`.
209 """
210 return response, metadata
212 def pre_list_group_stats(
213 self,
214 request: error_stats_service.ListGroupStatsRequest,
215 metadata: Sequence[Tuple[str, Union[str, bytes]]],
216 ) -> Tuple[
217 error_stats_service.ListGroupStatsRequest,
218 Sequence[Tuple[str, Union[str, bytes]]],
219 ]:
220 """Pre-rpc interceptor for list_group_stats
222 Override in a subclass to manipulate the request or metadata
223 before they are sent to the ErrorStatsService server.
224 """
225 return request, metadata
227 def post_list_group_stats(
228 self, response: error_stats_service.ListGroupStatsResponse
229 ) -> error_stats_service.ListGroupStatsResponse:
230 """Post-rpc interceptor for list_group_stats
232 DEPRECATED. Please use the `post_list_group_stats_with_metadata`
233 interceptor instead.
235 Override in a subclass to read or manipulate the response
236 after it is returned by the ErrorStatsService server but before
237 it is returned to user code. This `post_list_group_stats` interceptor runs
238 before the `post_list_group_stats_with_metadata` interceptor.
239 """
240 return response
242 def post_list_group_stats_with_metadata(
243 self,
244 response: error_stats_service.ListGroupStatsResponse,
245 metadata: Sequence[Tuple[str, Union[str, bytes]]],
246 ) -> Tuple[
247 error_stats_service.ListGroupStatsResponse,
248 Sequence[Tuple[str, Union[str, bytes]]],
249 ]:
250 """Post-rpc interceptor for list_group_stats
252 Override in a subclass to read or manipulate the response or metadata after it
253 is returned by the ErrorStatsService server but before it is returned to user code.
255 We recommend only using this `post_list_group_stats_with_metadata`
256 interceptor in new development instead of the `post_list_group_stats` interceptor.
257 When both interceptors are used, this `post_list_group_stats_with_metadata` interceptor runs after the
258 `post_list_group_stats` interceptor. The (possibly modified) response returned by
259 `post_list_group_stats` will be passed to
260 `post_list_group_stats_with_metadata`.
261 """
262 return response, metadata
265@dataclasses.dataclass
266class ErrorStatsServiceRestStub:
267 _session: AuthorizedSession
268 _host: str
269 _interceptor: ErrorStatsServiceRestInterceptor
272class ErrorStatsServiceRestTransport(_BaseErrorStatsServiceRestTransport):
273 """REST backend synchronous transport for ErrorStatsService.
275 An API for retrieving and managing error statistics as well
276 as data for individual events.
278 This class defines the same methods as the primary client, so the
279 primary client can load the underlying transport implementation
280 and call it.
282 It sends JSON representations of protocol buffers over HTTP/1.1
283 """
285 def __init__(
286 self,
287 *,
288 host: str = "clouderrorreporting.googleapis.com",
289 credentials: Optional[ga_credentials.Credentials] = None,
290 credentials_file: Optional[str] = None,
291 scopes: Optional[Sequence[str]] = None,
292 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
293 quota_project_id: Optional[str] = None,
294 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
295 always_use_jwt_access: Optional[bool] = False,
296 url_scheme: str = "https",
297 interceptor: Optional[ErrorStatsServiceRestInterceptor] = None,
298 api_audience: Optional[str] = None,
299 ) -> None:
300 """Instantiate the transport.
302 Args:
303 host (Optional[str]):
304 The hostname to connect to (default: 'clouderrorreporting.googleapis.com').
305 credentials (Optional[google.auth.credentials.Credentials]): The
306 authorization credentials to attach to requests. These
307 credentials identify the application to the service; if none
308 are specified, the client will attempt to ascertain the
309 credentials from the environment.
311 credentials_file (Optional[str]): A file with credentials that can
312 be loaded with :func:`google.auth.load_credentials_from_file`.
313 This argument is ignored if ``channel`` is provided.
314 scopes (Optional(Sequence[str])): A list of scopes. This argument is
315 ignored if ``channel`` is provided.
316 client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
317 certificate to configure mutual TLS HTTP channel. It is ignored
318 if ``channel`` is provided.
319 quota_project_id (Optional[str]): An optional project to use for billing
320 and quota.
321 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
322 The client info used to send a user-agent string along with
323 API requests. If ``None``, then default info will be used.
324 Generally, you only need to set this if you are developing
325 your own client library.
326 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
327 be used for service account credentials.
328 url_scheme: the protocol scheme for the API endpoint. Normally
329 "https", but for testing or local servers,
330 "http" can be specified.
331 """
332 # Run the base constructor
333 # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
334 # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
335 # credentials object
336 super().__init__(
337 host=host,
338 credentials=credentials,
339 client_info=client_info,
340 always_use_jwt_access=always_use_jwt_access,
341 url_scheme=url_scheme,
342 api_audience=api_audience,
343 )
344 self._session = AuthorizedSession(
345 self._credentials, default_host=self.DEFAULT_HOST
346 )
347 if client_cert_source_for_mtls:
348 self._session.configure_mtls_channel(client_cert_source_for_mtls)
349 self._interceptor = interceptor or ErrorStatsServiceRestInterceptor()
350 self._prep_wrapped_messages(client_info)
352 class _DeleteEvents(
353 _BaseErrorStatsServiceRestTransport._BaseDeleteEvents, ErrorStatsServiceRestStub
354 ):
355 def __hash__(self):
356 return hash("ErrorStatsServiceRestTransport.DeleteEvents")
358 @staticmethod
359 def _get_response(
360 host,
361 metadata,
362 query_params,
363 session,
364 timeout,
365 transcoded_request,
366 body=None,
367 ):
368 uri = transcoded_request["uri"]
369 method = transcoded_request["method"]
370 headers = dict(metadata)
371 headers["Content-Type"] = "application/json"
372 response = getattr(session, method)(
373 "{host}{uri}".format(host=host, uri=uri),
374 timeout=timeout,
375 headers=headers,
376 params=rest_helpers.flatten_query_params(query_params, strict=True),
377 )
378 return response
380 def __call__(
381 self,
382 request: error_stats_service.DeleteEventsRequest,
383 *,
384 retry: OptionalRetry = gapic_v1.method.DEFAULT,
385 timeout: Optional[float] = None,
386 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
387 ) -> error_stats_service.DeleteEventsResponse:
388 r"""Call the delete events method over HTTP.
390 Args:
391 request (~.error_stats_service.DeleteEventsRequest):
392 The request object. Deletes all events in the project.
393 retry (google.api_core.retry.Retry): Designation of what errors, if any,
394 should be retried.
395 timeout (float): The timeout for this request.
396 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
397 sent along with the request as metadata. Normally, each value must be of type `str`,
398 but for metadata keys ending with the suffix `-bin`, the corresponding values must
399 be of type `bytes`.
401 Returns:
402 ~.error_stats_service.DeleteEventsResponse:
403 Response message for deleting error
404 events.
406 """
408 http_options = (
409 _BaseErrorStatsServiceRestTransport._BaseDeleteEvents._get_http_options()
410 )
412 request, metadata = self._interceptor.pre_delete_events(request, metadata)
413 transcoded_request = _BaseErrorStatsServiceRestTransport._BaseDeleteEvents._get_transcoded_request(
414 http_options, request
415 )
417 # Jsonify the query params
418 query_params = _BaseErrorStatsServiceRestTransport._BaseDeleteEvents._get_query_params_json(
419 transcoded_request
420 )
422 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
423 logging.DEBUG
424 ): # pragma: NO COVER
425 request_url = "{host}{uri}".format(
426 host=self._host, uri=transcoded_request["uri"]
427 )
428 method = transcoded_request["method"]
429 try:
430 request_payload = type(request).to_json(request)
431 except:
432 request_payload = None
433 http_request = {
434 "payload": request_payload,
435 "requestMethod": method,
436 "requestUrl": request_url,
437 "headers": dict(metadata),
438 }
439 _LOGGER.debug(
440 f"Sending request for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.DeleteEvents",
441 extra={
442 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
443 "rpcName": "DeleteEvents",
444 "httpRequest": http_request,
445 "metadata": http_request["headers"],
446 },
447 )
449 # Send the request
450 response = ErrorStatsServiceRestTransport._DeleteEvents._get_response(
451 self._host,
452 metadata,
453 query_params,
454 self._session,
455 timeout,
456 transcoded_request,
457 )
459 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
460 # subclass.
461 if response.status_code >= 400:
462 raise core_exceptions.from_http_response(response)
464 # Return the response
465 resp = error_stats_service.DeleteEventsResponse()
466 pb_resp = error_stats_service.DeleteEventsResponse.pb(resp)
468 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
470 resp = self._interceptor.post_delete_events(resp)
471 response_metadata = [(k, str(v)) for k, v in response.headers.items()]
472 resp, _ = self._interceptor.post_delete_events_with_metadata(
473 resp, response_metadata
474 )
475 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
476 logging.DEBUG
477 ): # pragma: NO COVER
478 try:
479 response_payload = error_stats_service.DeleteEventsResponse.to_json(
480 response
481 )
482 except:
483 response_payload = None
484 http_response = {
485 "payload": response_payload,
486 "headers": dict(response.headers),
487 "status": response.status_code,
488 }
489 _LOGGER.debug(
490 "Received response for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.delete_events",
491 extra={
492 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
493 "rpcName": "DeleteEvents",
494 "metadata": http_response["headers"],
495 "httpResponse": http_response,
496 },
497 )
498 return resp
500 class _ListEvents(
501 _BaseErrorStatsServiceRestTransport._BaseListEvents, ErrorStatsServiceRestStub
502 ):
503 def __hash__(self):
504 return hash("ErrorStatsServiceRestTransport.ListEvents")
506 @staticmethod
507 def _get_response(
508 host,
509 metadata,
510 query_params,
511 session,
512 timeout,
513 transcoded_request,
514 body=None,
515 ):
516 uri = transcoded_request["uri"]
517 method = transcoded_request["method"]
518 headers = dict(metadata)
519 headers["Content-Type"] = "application/json"
520 response = getattr(session, method)(
521 "{host}{uri}".format(host=host, uri=uri),
522 timeout=timeout,
523 headers=headers,
524 params=rest_helpers.flatten_query_params(query_params, strict=True),
525 )
526 return response
528 def __call__(
529 self,
530 request: error_stats_service.ListEventsRequest,
531 *,
532 retry: OptionalRetry = gapic_v1.method.DEFAULT,
533 timeout: Optional[float] = None,
534 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
535 ) -> error_stats_service.ListEventsResponse:
536 r"""Call the list events method over HTTP.
538 Args:
539 request (~.error_stats_service.ListEventsRequest):
540 The request object. Specifies a set of error events to
541 return.
542 retry (google.api_core.retry.Retry): Designation of what errors, if any,
543 should be retried.
544 timeout (float): The timeout for this request.
545 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
546 sent along with the request as metadata. Normally, each value must be of type `str`,
547 but for metadata keys ending with the suffix `-bin`, the corresponding values must
548 be of type `bytes`.
550 Returns:
551 ~.error_stats_service.ListEventsResponse:
552 Contains a set of requested error
553 events.
555 """
557 http_options = (
558 _BaseErrorStatsServiceRestTransport._BaseListEvents._get_http_options()
559 )
561 request, metadata = self._interceptor.pre_list_events(request, metadata)
562 transcoded_request = _BaseErrorStatsServiceRestTransport._BaseListEvents._get_transcoded_request(
563 http_options, request
564 )
566 # Jsonify the query params
567 query_params = _BaseErrorStatsServiceRestTransport._BaseListEvents._get_query_params_json(
568 transcoded_request
569 )
571 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
572 logging.DEBUG
573 ): # pragma: NO COVER
574 request_url = "{host}{uri}".format(
575 host=self._host, uri=transcoded_request["uri"]
576 )
577 method = transcoded_request["method"]
578 try:
579 request_payload = type(request).to_json(request)
580 except:
581 request_payload = None
582 http_request = {
583 "payload": request_payload,
584 "requestMethod": method,
585 "requestUrl": request_url,
586 "headers": dict(metadata),
587 }
588 _LOGGER.debug(
589 f"Sending request for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.ListEvents",
590 extra={
591 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
592 "rpcName": "ListEvents",
593 "httpRequest": http_request,
594 "metadata": http_request["headers"],
595 },
596 )
598 # Send the request
599 response = ErrorStatsServiceRestTransport._ListEvents._get_response(
600 self._host,
601 metadata,
602 query_params,
603 self._session,
604 timeout,
605 transcoded_request,
606 )
608 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
609 # subclass.
610 if response.status_code >= 400:
611 raise core_exceptions.from_http_response(response)
613 # Return the response
614 resp = error_stats_service.ListEventsResponse()
615 pb_resp = error_stats_service.ListEventsResponse.pb(resp)
617 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
619 resp = self._interceptor.post_list_events(resp)
620 response_metadata = [(k, str(v)) for k, v in response.headers.items()]
621 resp, _ = self._interceptor.post_list_events_with_metadata(
622 resp, response_metadata
623 )
624 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
625 logging.DEBUG
626 ): # pragma: NO COVER
627 try:
628 response_payload = error_stats_service.ListEventsResponse.to_json(
629 response
630 )
631 except:
632 response_payload = None
633 http_response = {
634 "payload": response_payload,
635 "headers": dict(response.headers),
636 "status": response.status_code,
637 }
638 _LOGGER.debug(
639 "Received response for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.list_events",
640 extra={
641 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
642 "rpcName": "ListEvents",
643 "metadata": http_response["headers"],
644 "httpResponse": http_response,
645 },
646 )
647 return resp
649 class _ListGroupStats(
650 _BaseErrorStatsServiceRestTransport._BaseListGroupStats,
651 ErrorStatsServiceRestStub,
652 ):
653 def __hash__(self):
654 return hash("ErrorStatsServiceRestTransport.ListGroupStats")
656 @staticmethod
657 def _get_response(
658 host,
659 metadata,
660 query_params,
661 session,
662 timeout,
663 transcoded_request,
664 body=None,
665 ):
666 uri = transcoded_request["uri"]
667 method = transcoded_request["method"]
668 headers = dict(metadata)
669 headers["Content-Type"] = "application/json"
670 response = getattr(session, method)(
671 "{host}{uri}".format(host=host, uri=uri),
672 timeout=timeout,
673 headers=headers,
674 params=rest_helpers.flatten_query_params(query_params, strict=True),
675 )
676 return response
678 def __call__(
679 self,
680 request: error_stats_service.ListGroupStatsRequest,
681 *,
682 retry: OptionalRetry = gapic_v1.method.DEFAULT,
683 timeout: Optional[float] = None,
684 metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
685 ) -> error_stats_service.ListGroupStatsResponse:
686 r"""Call the list group stats method over HTTP.
688 Args:
689 request (~.error_stats_service.ListGroupStatsRequest):
690 The request object. Specifies a set of ``ErrorGroupStats`` to return.
691 retry (google.api_core.retry.Retry): Designation of what errors, if any,
692 should be retried.
693 timeout (float): The timeout for this request.
694 metadata (Sequence[Tuple[str, Union[str, bytes]]]): Key/value pairs which should be
695 sent along with the request as metadata. Normally, each value must be of type `str`,
696 but for metadata keys ending with the suffix `-bin`, the corresponding values must
697 be of type `bytes`.
699 Returns:
700 ~.error_stats_service.ListGroupStatsResponse:
701 Contains a set of requested error
702 group stats.
704 """
706 http_options = (
707 _BaseErrorStatsServiceRestTransport._BaseListGroupStats._get_http_options()
708 )
710 request, metadata = self._interceptor.pre_list_group_stats(
711 request, metadata
712 )
713 transcoded_request = _BaseErrorStatsServiceRestTransport._BaseListGroupStats._get_transcoded_request(
714 http_options, request
715 )
717 # Jsonify the query params
718 query_params = _BaseErrorStatsServiceRestTransport._BaseListGroupStats._get_query_params_json(
719 transcoded_request
720 )
722 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
723 logging.DEBUG
724 ): # pragma: NO COVER
725 request_url = "{host}{uri}".format(
726 host=self._host, uri=transcoded_request["uri"]
727 )
728 method = transcoded_request["method"]
729 try:
730 request_payload = type(request).to_json(request)
731 except:
732 request_payload = None
733 http_request = {
734 "payload": request_payload,
735 "requestMethod": method,
736 "requestUrl": request_url,
737 "headers": dict(metadata),
738 }
739 _LOGGER.debug(
740 f"Sending request for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.ListGroupStats",
741 extra={
742 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
743 "rpcName": "ListGroupStats",
744 "httpRequest": http_request,
745 "metadata": http_request["headers"],
746 },
747 )
749 # Send the request
750 response = ErrorStatsServiceRestTransport._ListGroupStats._get_response(
751 self._host,
752 metadata,
753 query_params,
754 self._session,
755 timeout,
756 transcoded_request,
757 )
759 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
760 # subclass.
761 if response.status_code >= 400:
762 raise core_exceptions.from_http_response(response)
764 # Return the response
765 resp = error_stats_service.ListGroupStatsResponse()
766 pb_resp = error_stats_service.ListGroupStatsResponse.pb(resp)
768 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
770 resp = self._interceptor.post_list_group_stats(resp)
771 response_metadata = [(k, str(v)) for k, v in response.headers.items()]
772 resp, _ = self._interceptor.post_list_group_stats_with_metadata(
773 resp, response_metadata
774 )
775 if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
776 logging.DEBUG
777 ): # pragma: NO COVER
778 try:
779 response_payload = (
780 error_stats_service.ListGroupStatsResponse.to_json(response)
781 )
782 except:
783 response_payload = None
784 http_response = {
785 "payload": response_payload,
786 "headers": dict(response.headers),
787 "status": response.status_code,
788 }
789 _LOGGER.debug(
790 "Received response for google.devtools.clouderrorreporting_v1beta1.ErrorStatsServiceClient.list_group_stats",
791 extra={
792 "serviceName": "google.devtools.clouderrorreporting.v1beta1.ErrorStatsService",
793 "rpcName": "ListGroupStats",
794 "metadata": http_response["headers"],
795 "httpResponse": http_response,
796 },
797 )
798 return resp
800 @property
801 def delete_events(
802 self,
803 ) -> Callable[
804 [error_stats_service.DeleteEventsRequest],
805 error_stats_service.DeleteEventsResponse,
806 ]:
807 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
808 # In C++ this would require a dynamic_cast
809 return self._DeleteEvents(self._session, self._host, self._interceptor) # type: ignore
811 @property
812 def list_events(
813 self,
814 ) -> Callable[
815 [error_stats_service.ListEventsRequest], error_stats_service.ListEventsResponse
816 ]:
817 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
818 # In C++ this would require a dynamic_cast
819 return self._ListEvents(self._session, self._host, self._interceptor) # type: ignore
821 @property
822 def list_group_stats(
823 self,
824 ) -> Callable[
825 [error_stats_service.ListGroupStatsRequest],
826 error_stats_service.ListGroupStatsResponse,
827 ]:
828 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
829 # In C++ this would require a dynamic_cast
830 return self._ListGroupStats(self._session, self._host, self._interceptor) # type: ignore
832 @property
833 def kind(self) -> str:
834 return "rest"
836 def close(self):
837 self._session.close()
840__all__ = ("ErrorStatsServiceRestTransport",)