« prev ^ index » next coverage.py v7.2.7, created at 2023-06-06 06:03 +0000
1# -*- coding: utf-8 -*-
2# Copyright 2022 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#
17import dataclasses
18import json # type: ignore
19import re
20from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
21import warnings
23from google.api_core import (
24 gapic_v1,
25 operations_v1,
26 path_template,
27 rest_helpers,
28 rest_streaming,
29)
30from google.api_core import exceptions as core_exceptions
31from google.api_core import retry as retries
32from google.auth import credentials as ga_credentials # type: ignore
33from google.auth.transport.grpc import SslCredentials # type: ignore
34from google.auth.transport.requests import AuthorizedSession # type: ignore
35from google.longrunning import operations_pb2
36from google.protobuf import json_format
37import grpc # type: ignore
38from requests import __version__ as requests_version
40try:
41 OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault]
42except AttributeError: # pragma: NO COVER
43 OptionalRetry = Union[retries.Retry, object] # type: ignore
46from google.longrunning import operations_pb2 # type: ignore
48from google.cloud.resourcemanager_v3.types import tag_holds
50from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO
51from .base import TagHoldsTransport
53DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
54 gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
55 grpc_version=None,
56 rest_version=requests_version,
57)
60class TagHoldsRestInterceptor:
61 """Interceptor for TagHolds.
63 Interceptors are used to manipulate requests, request metadata, and responses
64 in arbitrary ways.
65 Example use cases include:
66 * Logging
67 * Verifying requests according to service or custom semantics
68 * Stripping extraneous information from responses
70 These use cases and more can be enabled by injecting an
71 instance of a custom subclass when constructing the TagHoldsRestTransport.
73 .. code-block:: python
74 class MyCustomTagHoldsInterceptor(TagHoldsRestInterceptor):
75 def pre_create_tag_hold(self, request, metadata):
76 logging.log(f"Received request: {request}")
77 return request, metadata
79 def post_create_tag_hold(self, response):
80 logging.log(f"Received response: {response}")
81 return response
83 def pre_delete_tag_hold(self, request, metadata):
84 logging.log(f"Received request: {request}")
85 return request, metadata
87 def post_delete_tag_hold(self, response):
88 logging.log(f"Received response: {response}")
89 return response
91 def pre_list_tag_holds(self, request, metadata):
92 logging.log(f"Received request: {request}")
93 return request, metadata
95 def post_list_tag_holds(self, response):
96 logging.log(f"Received response: {response}")
97 return response
99 transport = TagHoldsRestTransport(interceptor=MyCustomTagHoldsInterceptor())
100 client = TagHoldsClient(transport=transport)
103 """
105 def pre_create_tag_hold(
106 self,
107 request: tag_holds.CreateTagHoldRequest,
108 metadata: Sequence[Tuple[str, str]],
109 ) -> Tuple[tag_holds.CreateTagHoldRequest, Sequence[Tuple[str, str]]]:
110 """Pre-rpc interceptor for create_tag_hold
112 Override in a subclass to manipulate the request or metadata
113 before they are sent to the TagHolds server.
114 """
115 return request, metadata
117 def post_create_tag_hold(
118 self, response: operations_pb2.Operation
119 ) -> operations_pb2.Operation:
120 """Post-rpc interceptor for create_tag_hold
122 Override in a subclass to manipulate the response
123 after it is returned by the TagHolds server but before
124 it is returned to user code.
125 """
126 return response
128 def pre_delete_tag_hold(
129 self,
130 request: tag_holds.DeleteTagHoldRequest,
131 metadata: Sequence[Tuple[str, str]],
132 ) -> Tuple[tag_holds.DeleteTagHoldRequest, Sequence[Tuple[str, str]]]:
133 """Pre-rpc interceptor for delete_tag_hold
135 Override in a subclass to manipulate the request or metadata
136 before they are sent to the TagHolds server.
137 """
138 return request, metadata
140 def post_delete_tag_hold(
141 self, response: operations_pb2.Operation
142 ) -> operations_pb2.Operation:
143 """Post-rpc interceptor for delete_tag_hold
145 Override in a subclass to manipulate the response
146 after it is returned by the TagHolds server but before
147 it is returned to user code.
148 """
149 return response
151 def pre_list_tag_holds(
152 self,
153 request: tag_holds.ListTagHoldsRequest,
154 metadata: Sequence[Tuple[str, str]],
155 ) -> Tuple[tag_holds.ListTagHoldsRequest, Sequence[Tuple[str, str]]]:
156 """Pre-rpc interceptor for list_tag_holds
158 Override in a subclass to manipulate the request or metadata
159 before they are sent to the TagHolds server.
160 """
161 return request, metadata
163 def post_list_tag_holds(
164 self, response: tag_holds.ListTagHoldsResponse
165 ) -> tag_holds.ListTagHoldsResponse:
166 """Post-rpc interceptor for list_tag_holds
168 Override in a subclass to manipulate the response
169 after it is returned by the TagHolds server but before
170 it is returned to user code.
171 """
172 return response
174 def pre_get_operation(
175 self,
176 request: operations_pb2.GetOperationRequest,
177 metadata: Sequence[Tuple[str, str]],
178 ) -> Tuple[operations_pb2.GetOperationRequest, Sequence[Tuple[str, str]]]:
179 """Pre-rpc interceptor for get_operation
181 Override in a subclass to manipulate the request or metadata
182 before they are sent to the TagHolds server.
183 """
184 return request, metadata
186 def post_get_operation(
187 self, response: operations_pb2.Operation
188 ) -> operations_pb2.Operation:
189 """Post-rpc interceptor for get_operation
191 Override in a subclass to manipulate the response
192 after it is returned by the TagHolds server but before
193 it is returned to user code.
194 """
195 return response
198@dataclasses.dataclass
199class TagHoldsRestStub:
200 _session: AuthorizedSession
201 _host: str
202 _interceptor: TagHoldsRestInterceptor
205class TagHoldsRestTransport(TagHoldsTransport):
206 """REST backend transport for TagHolds.
208 Allow users to create and manage TagHolds for TagValues.
209 TagHolds represent the use of a Tag Value that is not captured
210 by TagBindings but should still block TagValue deletion (such as
211 a reference in a policy condition). This service provides
212 isolated failure domains by cloud location so that TagHolds can
213 be managed in the same location as their usage.
215 This class defines the same methods as the primary client, so the
216 primary client can load the underlying transport implementation
217 and call it.
219 It sends JSON representations of protocol buffers over HTTP/1.1
221 """
223 def __init__(
224 self,
225 *,
226 host: str = "cloudresourcemanager.googleapis.com",
227 credentials: Optional[ga_credentials.Credentials] = None,
228 credentials_file: Optional[str] = None,
229 scopes: Optional[Sequence[str]] = None,
230 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
231 quota_project_id: Optional[str] = None,
232 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
233 always_use_jwt_access: Optional[bool] = False,
234 url_scheme: str = "https",
235 interceptor: Optional[TagHoldsRestInterceptor] = None,
236 api_audience: Optional[str] = None,
237 ) -> None:
238 """Instantiate the transport.
240 Args:
241 host (Optional[str]):
242 The hostname to connect to.
243 credentials (Optional[google.auth.credentials.Credentials]): The
244 authorization credentials to attach to requests. These
245 credentials identify the application to the service; if none
246 are specified, the client will attempt to ascertain the
247 credentials from the environment.
249 credentials_file (Optional[str]): A file with credentials that can
250 be loaded with :func:`google.auth.load_credentials_from_file`.
251 This argument is ignored if ``channel`` is provided.
252 scopes (Optional(Sequence[str])): A list of scopes. This argument is
253 ignored if ``channel`` is provided.
254 client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
255 certificate to configure mutual TLS HTTP channel. It is ignored
256 if ``channel`` is provided.
257 quota_project_id (Optional[str]): An optional project to use for billing
258 and quota.
259 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
260 The client info used to send a user-agent string along with
261 API requests. If ``None``, then default info will be used.
262 Generally, you only need to set this if you are developing
263 your own client library.
264 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
265 be used for service account credentials.
266 url_scheme: the protocol scheme for the API endpoint. Normally
267 "https", but for testing or local servers,
268 "http" can be specified.
269 """
270 # Run the base constructor
271 # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
272 # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
273 # credentials object
274 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
275 if maybe_url_match is None:
276 raise ValueError(
277 f"Unexpected hostname structure: {host}"
278 ) # pragma: NO COVER
280 url_match_items = maybe_url_match.groupdict()
282 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
284 super().__init__(
285 host=host,
286 credentials=credentials,
287 client_info=client_info,
288 always_use_jwt_access=always_use_jwt_access,
289 api_audience=api_audience,
290 )
291 self._session = AuthorizedSession(
292 self._credentials, default_host=self.DEFAULT_HOST
293 )
294 self._operations_client: Optional[operations_v1.AbstractOperationsClient] = None
295 if client_cert_source_for_mtls:
296 self._session.configure_mtls_channel(client_cert_source_for_mtls)
297 self._interceptor = interceptor or TagHoldsRestInterceptor()
298 self._prep_wrapped_messages(client_info)
300 @property
301 def operations_client(self) -> operations_v1.AbstractOperationsClient:
302 """Create the client designed to process long-running operations.
304 This property caches on the instance; repeated calls return the same
305 client.
306 """
307 # Only create a new client if we do not already have one.
308 if self._operations_client is None:
309 http_options: Dict[str, List[Dict[str, str]]] = {
310 "google.longrunning.Operations.GetOperation": [
311 {
312 "method": "get",
313 "uri": "/v3/{name=operations/**}",
314 },
315 ],
316 }
318 rest_transport = operations_v1.OperationsRestTransport(
319 host=self._host,
320 # use the credentials which are saved
321 credentials=self._credentials,
322 scopes=self._scopes,
323 http_options=http_options,
324 path_prefix="v3",
325 )
327 self._operations_client = operations_v1.AbstractOperationsClient(
328 transport=rest_transport
329 )
331 # Return the client from cache.
332 return self._operations_client
334 class _CreateTagHold(TagHoldsRestStub):
335 def __hash__(self):
336 return hash("CreateTagHold")
338 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
340 @classmethod
341 def _get_unset_required_fields(cls, message_dict):
342 return {
343 k: v
344 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
345 if k not in message_dict
346 }
348 def __call__(
349 self,
350 request: tag_holds.CreateTagHoldRequest,
351 *,
352 retry: OptionalRetry = gapic_v1.method.DEFAULT,
353 timeout: Optional[float] = None,
354 metadata: Sequence[Tuple[str, str]] = (),
355 ) -> operations_pb2.Operation:
356 r"""Call the create tag hold method over HTTP.
358 Args:
359 request (~.tag_holds.CreateTagHoldRequest):
360 The request object. The request message to create a
361 TagHold.
362 retry (google.api_core.retry.Retry): Designation of what errors, if any,
363 should be retried.
364 timeout (float): The timeout for this request.
365 metadata (Sequence[Tuple[str, str]]): Strings which should be
366 sent along with the request as metadata.
368 Returns:
369 ~.operations_pb2.Operation:
370 This resource represents a
371 long-running operation that is the
372 result of a network API call.
374 """
376 http_options: List[Dict[str, str]] = [
377 {
378 "method": "post",
379 "uri": "/v3/{parent=tagValues/*}/tagHolds",
380 "body": "tag_hold",
381 },
382 ]
383 request, metadata = self._interceptor.pre_create_tag_hold(request, metadata)
384 pb_request = tag_holds.CreateTagHoldRequest.pb(request)
385 transcoded_request = path_template.transcode(http_options, pb_request)
387 # Jsonify the request body
389 body = json_format.MessageToJson(
390 transcoded_request["body"],
391 including_default_value_fields=False,
392 use_integers_for_enums=True,
393 )
394 uri = transcoded_request["uri"]
395 method = transcoded_request["method"]
397 # Jsonify the query params
398 query_params = json.loads(
399 json_format.MessageToJson(
400 transcoded_request["query_params"],
401 including_default_value_fields=False,
402 use_integers_for_enums=True,
403 )
404 )
405 query_params.update(self._get_unset_required_fields(query_params))
407 query_params["$alt"] = "json;enum-encoding=int"
409 # Send the request
410 headers = dict(metadata)
411 headers["Content-Type"] = "application/json"
412 response = getattr(self._session, method)(
413 "{host}{uri}".format(host=self._host, uri=uri),
414 timeout=timeout,
415 headers=headers,
416 params=rest_helpers.flatten_query_params(query_params, strict=True),
417 data=body,
418 )
420 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
421 # subclass.
422 if response.status_code >= 400:
423 raise core_exceptions.from_http_response(response)
425 # Return the response
426 resp = operations_pb2.Operation()
427 json_format.Parse(response.content, resp, ignore_unknown_fields=True)
428 resp = self._interceptor.post_create_tag_hold(resp)
429 return resp
431 class _DeleteTagHold(TagHoldsRestStub):
432 def __hash__(self):
433 return hash("DeleteTagHold")
435 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
437 @classmethod
438 def _get_unset_required_fields(cls, message_dict):
439 return {
440 k: v
441 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
442 if k not in message_dict
443 }
445 def __call__(
446 self,
447 request: tag_holds.DeleteTagHoldRequest,
448 *,
449 retry: OptionalRetry = gapic_v1.method.DEFAULT,
450 timeout: Optional[float] = None,
451 metadata: Sequence[Tuple[str, str]] = (),
452 ) -> operations_pb2.Operation:
453 r"""Call the delete tag hold method over HTTP.
455 Args:
456 request (~.tag_holds.DeleteTagHoldRequest):
457 The request object. The request message to delete a
458 TagHold.
459 retry (google.api_core.retry.Retry): Designation of what errors, if any,
460 should be retried.
461 timeout (float): The timeout for this request.
462 metadata (Sequence[Tuple[str, str]]): Strings which should be
463 sent along with the request as metadata.
465 Returns:
466 ~.operations_pb2.Operation:
467 This resource represents a
468 long-running operation that is the
469 result of a network API call.
471 """
473 http_options: List[Dict[str, str]] = [
474 {
475 "method": "delete",
476 "uri": "/v3/{name=tagValues/*/tagHolds/*}",
477 },
478 ]
479 request, metadata = self._interceptor.pre_delete_tag_hold(request, metadata)
480 pb_request = tag_holds.DeleteTagHoldRequest.pb(request)
481 transcoded_request = path_template.transcode(http_options, pb_request)
483 uri = transcoded_request["uri"]
484 method = transcoded_request["method"]
486 # Jsonify the query params
487 query_params = json.loads(
488 json_format.MessageToJson(
489 transcoded_request["query_params"],
490 including_default_value_fields=False,
491 use_integers_for_enums=True,
492 )
493 )
494 query_params.update(self._get_unset_required_fields(query_params))
496 query_params["$alt"] = "json;enum-encoding=int"
498 # Send the request
499 headers = dict(metadata)
500 headers["Content-Type"] = "application/json"
501 response = getattr(self._session, method)(
502 "{host}{uri}".format(host=self._host, uri=uri),
503 timeout=timeout,
504 headers=headers,
505 params=rest_helpers.flatten_query_params(query_params, strict=True),
506 )
508 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
509 # subclass.
510 if response.status_code >= 400:
511 raise core_exceptions.from_http_response(response)
513 # Return the response
514 resp = operations_pb2.Operation()
515 json_format.Parse(response.content, resp, ignore_unknown_fields=True)
516 resp = self._interceptor.post_delete_tag_hold(resp)
517 return resp
519 class _ListTagHolds(TagHoldsRestStub):
520 def __hash__(self):
521 return hash("ListTagHolds")
523 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
525 @classmethod
526 def _get_unset_required_fields(cls, message_dict):
527 return {
528 k: v
529 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
530 if k not in message_dict
531 }
533 def __call__(
534 self,
535 request: tag_holds.ListTagHoldsRequest,
536 *,
537 retry: OptionalRetry = gapic_v1.method.DEFAULT,
538 timeout: Optional[float] = None,
539 metadata: Sequence[Tuple[str, str]] = (),
540 ) -> tag_holds.ListTagHoldsResponse:
541 r"""Call the list tag holds method over HTTP.
543 Args:
544 request (~.tag_holds.ListTagHoldsRequest):
545 The request object. The request message for listing the
546 TagHolds under a TagValue.
547 retry (google.api_core.retry.Retry): Designation of what errors, if any,
548 should be retried.
549 timeout (float): The timeout for this request.
550 metadata (Sequence[Tuple[str, str]]): Strings which should be
551 sent along with the request as metadata.
553 Returns:
554 ~.tag_holds.ListTagHoldsResponse:
555 The ListTagHolds response.
556 """
558 http_options: List[Dict[str, str]] = [
559 {
560 "method": "get",
561 "uri": "/v3/{parent=tagValues/*}/tagHolds",
562 },
563 ]
564 request, metadata = self._interceptor.pre_list_tag_holds(request, metadata)
565 pb_request = tag_holds.ListTagHoldsRequest.pb(request)
566 transcoded_request = path_template.transcode(http_options, pb_request)
568 uri = transcoded_request["uri"]
569 method = transcoded_request["method"]
571 # Jsonify the query params
572 query_params = json.loads(
573 json_format.MessageToJson(
574 transcoded_request["query_params"],
575 including_default_value_fields=False,
576 use_integers_for_enums=True,
577 )
578 )
579 query_params.update(self._get_unset_required_fields(query_params))
581 query_params["$alt"] = "json;enum-encoding=int"
583 # Send the request
584 headers = dict(metadata)
585 headers["Content-Type"] = "application/json"
586 response = getattr(self._session, method)(
587 "{host}{uri}".format(host=self._host, uri=uri),
588 timeout=timeout,
589 headers=headers,
590 params=rest_helpers.flatten_query_params(query_params, strict=True),
591 )
593 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
594 # subclass.
595 if response.status_code >= 400:
596 raise core_exceptions.from_http_response(response)
598 # Return the response
599 resp = tag_holds.ListTagHoldsResponse()
600 pb_resp = tag_holds.ListTagHoldsResponse.pb(resp)
602 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
603 resp = self._interceptor.post_list_tag_holds(resp)
604 return resp
606 @property
607 def create_tag_hold(
608 self,
609 ) -> Callable[[tag_holds.CreateTagHoldRequest], operations_pb2.Operation]:
610 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
611 # In C++ this would require a dynamic_cast
612 return self._CreateTagHold(self._session, self._host, self._interceptor) # type: ignore
614 @property
615 def delete_tag_hold(
616 self,
617 ) -> Callable[[tag_holds.DeleteTagHoldRequest], operations_pb2.Operation]:
618 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
619 # In C++ this would require a dynamic_cast
620 return self._DeleteTagHold(self._session, self._host, self._interceptor) # type: ignore
622 @property
623 def list_tag_holds(
624 self,
625 ) -> Callable[[tag_holds.ListTagHoldsRequest], tag_holds.ListTagHoldsResponse]:
626 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
627 # In C++ this would require a dynamic_cast
628 return self._ListTagHolds(self._session, self._host, self._interceptor) # type: ignore
630 @property
631 def get_operation(self):
632 return self._GetOperation(self._session, self._host, self._interceptor) # type: ignore
634 class _GetOperation(TagHoldsRestStub):
635 def __call__(
636 self,
637 request: operations_pb2.GetOperationRequest,
638 *,
639 retry: OptionalRetry = gapic_v1.method.DEFAULT,
640 timeout: Optional[float] = None,
641 metadata: Sequence[Tuple[str, str]] = (),
642 ) -> operations_pb2.Operation:
644 r"""Call the get operation method over HTTP.
646 Args:
647 request (operations_pb2.GetOperationRequest):
648 The request object for GetOperation method.
649 retry (google.api_core.retry.Retry): Designation of what errors, if any,
650 should be retried.
651 timeout (float): The timeout for this request.
652 metadata (Sequence[Tuple[str, str]]): Strings which should be
653 sent along with the request as metadata.
655 Returns:
656 operations_pb2.Operation: Response from GetOperation method.
657 """
659 http_options: List[Dict[str, str]] = [
660 {
661 "method": "get",
662 "uri": "/v3/{name=operations/**}",
663 },
664 ]
666 request, metadata = self._interceptor.pre_get_operation(request, metadata)
667 request_kwargs = json_format.MessageToDict(request)
668 transcoded_request = path_template.transcode(http_options, **request_kwargs)
670 uri = transcoded_request["uri"]
671 method = transcoded_request["method"]
673 # Jsonify the query params
674 query_params = json.loads(json.dumps(transcoded_request["query_params"]))
676 # Send the request
677 headers = dict(metadata)
678 headers["Content-Type"] = "application/json"
680 response = getattr(self._session, method)(
681 "{host}{uri}".format(host=self._host, uri=uri),
682 timeout=timeout,
683 headers=headers,
684 params=rest_helpers.flatten_query_params(query_params),
685 )
687 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
688 # subclass.
689 if response.status_code >= 400:
690 raise core_exceptions.from_http_response(response)
692 resp = operations_pb2.Operation()
693 resp = json_format.Parse(response.content.decode("utf-8"), resp)
694 resp = self._interceptor.post_get_operation(resp)
695 return resp
697 @property
698 def kind(self) -> str:
699 return "rest"
701 def close(self):
702 self._session.close()
705__all__ = ("TagHoldsRestTransport",)