« 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_bindings
50from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO
51from .base import TagBindingsTransport
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 TagBindingsRestInterceptor:
61 """Interceptor for TagBindings.
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 TagBindingsRestTransport.
73 .. code-block:: python
74 class MyCustomTagBindingsInterceptor(TagBindingsRestInterceptor):
75 def pre_create_tag_binding(self, request, metadata):
76 logging.log(f"Received request: {request}")
77 return request, metadata
79 def post_create_tag_binding(self, response):
80 logging.log(f"Received response: {response}")
81 return response
83 def pre_delete_tag_binding(self, request, metadata):
84 logging.log(f"Received request: {request}")
85 return request, metadata
87 def post_delete_tag_binding(self, response):
88 logging.log(f"Received response: {response}")
89 return response
91 def pre_list_effective_tags(self, request, metadata):
92 logging.log(f"Received request: {request}")
93 return request, metadata
95 def post_list_effective_tags(self, response):
96 logging.log(f"Received response: {response}")
97 return response
99 def pre_list_tag_bindings(self, request, metadata):
100 logging.log(f"Received request: {request}")
101 return request, metadata
103 def post_list_tag_bindings(self, response):
104 logging.log(f"Received response: {response}")
105 return response
107 transport = TagBindingsRestTransport(interceptor=MyCustomTagBindingsInterceptor())
108 client = TagBindingsClient(transport=transport)
111 """
113 def pre_create_tag_binding(
114 self,
115 request: tag_bindings.CreateTagBindingRequest,
116 metadata: Sequence[Tuple[str, str]],
117 ) -> Tuple[tag_bindings.CreateTagBindingRequest, Sequence[Tuple[str, str]]]:
118 """Pre-rpc interceptor for create_tag_binding
120 Override in a subclass to manipulate the request or metadata
121 before they are sent to the TagBindings server.
122 """
123 return request, metadata
125 def post_create_tag_binding(
126 self, response: operations_pb2.Operation
127 ) -> operations_pb2.Operation:
128 """Post-rpc interceptor for create_tag_binding
130 Override in a subclass to manipulate the response
131 after it is returned by the TagBindings server but before
132 it is returned to user code.
133 """
134 return response
136 def pre_delete_tag_binding(
137 self,
138 request: tag_bindings.DeleteTagBindingRequest,
139 metadata: Sequence[Tuple[str, str]],
140 ) -> Tuple[tag_bindings.DeleteTagBindingRequest, Sequence[Tuple[str, str]]]:
141 """Pre-rpc interceptor for delete_tag_binding
143 Override in a subclass to manipulate the request or metadata
144 before they are sent to the TagBindings server.
145 """
146 return request, metadata
148 def post_delete_tag_binding(
149 self, response: operations_pb2.Operation
150 ) -> operations_pb2.Operation:
151 """Post-rpc interceptor for delete_tag_binding
153 Override in a subclass to manipulate the response
154 after it is returned by the TagBindings server but before
155 it is returned to user code.
156 """
157 return response
159 def pre_list_effective_tags(
160 self,
161 request: tag_bindings.ListEffectiveTagsRequest,
162 metadata: Sequence[Tuple[str, str]],
163 ) -> Tuple[tag_bindings.ListEffectiveTagsRequest, Sequence[Tuple[str, str]]]:
164 """Pre-rpc interceptor for list_effective_tags
166 Override in a subclass to manipulate the request or metadata
167 before they are sent to the TagBindings server.
168 """
169 return request, metadata
171 def post_list_effective_tags(
172 self, response: tag_bindings.ListEffectiveTagsResponse
173 ) -> tag_bindings.ListEffectiveTagsResponse:
174 """Post-rpc interceptor for list_effective_tags
176 Override in a subclass to manipulate the response
177 after it is returned by the TagBindings server but before
178 it is returned to user code.
179 """
180 return response
182 def pre_list_tag_bindings(
183 self,
184 request: tag_bindings.ListTagBindingsRequest,
185 metadata: Sequence[Tuple[str, str]],
186 ) -> Tuple[tag_bindings.ListTagBindingsRequest, Sequence[Tuple[str, str]]]:
187 """Pre-rpc interceptor for list_tag_bindings
189 Override in a subclass to manipulate the request or metadata
190 before they are sent to the TagBindings server.
191 """
192 return request, metadata
194 def post_list_tag_bindings(
195 self, response: tag_bindings.ListTagBindingsResponse
196 ) -> tag_bindings.ListTagBindingsResponse:
197 """Post-rpc interceptor for list_tag_bindings
199 Override in a subclass to manipulate the response
200 after it is returned by the TagBindings server but before
201 it is returned to user code.
202 """
203 return response
205 def pre_get_operation(
206 self,
207 request: operations_pb2.GetOperationRequest,
208 metadata: Sequence[Tuple[str, str]],
209 ) -> Tuple[operations_pb2.GetOperationRequest, Sequence[Tuple[str, str]]]:
210 """Pre-rpc interceptor for get_operation
212 Override in a subclass to manipulate the request or metadata
213 before they are sent to the TagBindings server.
214 """
215 return request, metadata
217 def post_get_operation(
218 self, response: operations_pb2.Operation
219 ) -> operations_pb2.Operation:
220 """Post-rpc interceptor for get_operation
222 Override in a subclass to manipulate the response
223 after it is returned by the TagBindings server but before
224 it is returned to user code.
225 """
226 return response
229@dataclasses.dataclass
230class TagBindingsRestStub:
231 _session: AuthorizedSession
232 _host: str
233 _interceptor: TagBindingsRestInterceptor
236class TagBindingsRestTransport(TagBindingsTransport):
237 """REST backend transport for TagBindings.
239 Allow users to create and manage TagBindings between
240 TagValues and different Google Cloud resources throughout the
241 GCP resource hierarchy.
243 This class defines the same methods as the primary client, so the
244 primary client can load the underlying transport implementation
245 and call it.
247 It sends JSON representations of protocol buffers over HTTP/1.1
249 """
251 def __init__(
252 self,
253 *,
254 host: str = "cloudresourcemanager.googleapis.com",
255 credentials: Optional[ga_credentials.Credentials] = None,
256 credentials_file: Optional[str] = None,
257 scopes: Optional[Sequence[str]] = None,
258 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
259 quota_project_id: Optional[str] = None,
260 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
261 always_use_jwt_access: Optional[bool] = False,
262 url_scheme: str = "https",
263 interceptor: Optional[TagBindingsRestInterceptor] = None,
264 api_audience: Optional[str] = None,
265 ) -> None:
266 """Instantiate the transport.
268 Args:
269 host (Optional[str]):
270 The hostname to connect to.
271 credentials (Optional[google.auth.credentials.Credentials]): The
272 authorization credentials to attach to requests. These
273 credentials identify the application to the service; if none
274 are specified, the client will attempt to ascertain the
275 credentials from the environment.
277 credentials_file (Optional[str]): A file with credentials that can
278 be loaded with :func:`google.auth.load_credentials_from_file`.
279 This argument is ignored if ``channel`` is provided.
280 scopes (Optional(Sequence[str])): A list of scopes. This argument is
281 ignored if ``channel`` is provided.
282 client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
283 certificate to configure mutual TLS HTTP channel. It is ignored
284 if ``channel`` is provided.
285 quota_project_id (Optional[str]): An optional project to use for billing
286 and quota.
287 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
288 The client info used to send a user-agent string along with
289 API requests. If ``None``, then default info will be used.
290 Generally, you only need to set this if you are developing
291 your own client library.
292 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
293 be used for service account credentials.
294 url_scheme: the protocol scheme for the API endpoint. Normally
295 "https", but for testing or local servers,
296 "http" can be specified.
297 """
298 # Run the base constructor
299 # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
300 # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
301 # credentials object
302 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
303 if maybe_url_match is None:
304 raise ValueError(
305 f"Unexpected hostname structure: {host}"
306 ) # pragma: NO COVER
308 url_match_items = maybe_url_match.groupdict()
310 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
312 super().__init__(
313 host=host,
314 credentials=credentials,
315 client_info=client_info,
316 always_use_jwt_access=always_use_jwt_access,
317 api_audience=api_audience,
318 )
319 self._session = AuthorizedSession(
320 self._credentials, default_host=self.DEFAULT_HOST
321 )
322 self._operations_client: Optional[operations_v1.AbstractOperationsClient] = None
323 if client_cert_source_for_mtls:
324 self._session.configure_mtls_channel(client_cert_source_for_mtls)
325 self._interceptor = interceptor or TagBindingsRestInterceptor()
326 self._prep_wrapped_messages(client_info)
328 @property
329 def operations_client(self) -> operations_v1.AbstractOperationsClient:
330 """Create the client designed to process long-running operations.
332 This property caches on the instance; repeated calls return the same
333 client.
334 """
335 # Only create a new client if we do not already have one.
336 if self._operations_client is None:
337 http_options: Dict[str, List[Dict[str, str]]] = {
338 "google.longrunning.Operations.GetOperation": [
339 {
340 "method": "get",
341 "uri": "/v3/{name=operations/**}",
342 },
343 ],
344 }
346 rest_transport = operations_v1.OperationsRestTransport(
347 host=self._host,
348 # use the credentials which are saved
349 credentials=self._credentials,
350 scopes=self._scopes,
351 http_options=http_options,
352 path_prefix="v3",
353 )
355 self._operations_client = operations_v1.AbstractOperationsClient(
356 transport=rest_transport
357 )
359 # Return the client from cache.
360 return self._operations_client
362 class _CreateTagBinding(TagBindingsRestStub):
363 def __hash__(self):
364 return hash("CreateTagBinding")
366 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
368 @classmethod
369 def _get_unset_required_fields(cls, message_dict):
370 return {
371 k: v
372 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
373 if k not in message_dict
374 }
376 def __call__(
377 self,
378 request: tag_bindings.CreateTagBindingRequest,
379 *,
380 retry: OptionalRetry = gapic_v1.method.DEFAULT,
381 timeout: Optional[float] = None,
382 metadata: Sequence[Tuple[str, str]] = (),
383 ) -> operations_pb2.Operation:
384 r"""Call the create tag binding method over HTTP.
386 Args:
387 request (~.tag_bindings.CreateTagBindingRequest):
388 The request object. The request message to create a
389 TagBinding.
390 retry (google.api_core.retry.Retry): Designation of what errors, if any,
391 should be retried.
392 timeout (float): The timeout for this request.
393 metadata (Sequence[Tuple[str, str]]): Strings which should be
394 sent along with the request as metadata.
396 Returns:
397 ~.operations_pb2.Operation:
398 This resource represents a
399 long-running operation that is the
400 result of a network API call.
402 """
404 http_options: List[Dict[str, str]] = [
405 {
406 "method": "post",
407 "uri": "/v3/tagBindings",
408 "body": "tag_binding",
409 },
410 ]
411 request, metadata = self._interceptor.pre_create_tag_binding(
412 request, metadata
413 )
414 pb_request = tag_bindings.CreateTagBindingRequest.pb(request)
415 transcoded_request = path_template.transcode(http_options, pb_request)
417 # Jsonify the request body
419 body = json_format.MessageToJson(
420 transcoded_request["body"],
421 including_default_value_fields=False,
422 use_integers_for_enums=True,
423 )
424 uri = transcoded_request["uri"]
425 method = transcoded_request["method"]
427 # Jsonify the query params
428 query_params = json.loads(
429 json_format.MessageToJson(
430 transcoded_request["query_params"],
431 including_default_value_fields=False,
432 use_integers_for_enums=True,
433 )
434 )
435 query_params.update(self._get_unset_required_fields(query_params))
437 query_params["$alt"] = "json;enum-encoding=int"
439 # Send the request
440 headers = dict(metadata)
441 headers["Content-Type"] = "application/json"
442 response = getattr(self._session, method)(
443 "{host}{uri}".format(host=self._host, uri=uri),
444 timeout=timeout,
445 headers=headers,
446 params=rest_helpers.flatten_query_params(query_params, strict=True),
447 data=body,
448 )
450 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
451 # subclass.
452 if response.status_code >= 400:
453 raise core_exceptions.from_http_response(response)
455 # Return the response
456 resp = operations_pb2.Operation()
457 json_format.Parse(response.content, resp, ignore_unknown_fields=True)
458 resp = self._interceptor.post_create_tag_binding(resp)
459 return resp
461 class _DeleteTagBinding(TagBindingsRestStub):
462 def __hash__(self):
463 return hash("DeleteTagBinding")
465 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {}
467 @classmethod
468 def _get_unset_required_fields(cls, message_dict):
469 return {
470 k: v
471 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
472 if k not in message_dict
473 }
475 def __call__(
476 self,
477 request: tag_bindings.DeleteTagBindingRequest,
478 *,
479 retry: OptionalRetry = gapic_v1.method.DEFAULT,
480 timeout: Optional[float] = None,
481 metadata: Sequence[Tuple[str, str]] = (),
482 ) -> operations_pb2.Operation:
483 r"""Call the delete tag binding method over HTTP.
485 Args:
486 request (~.tag_bindings.DeleteTagBindingRequest):
487 The request object. The request message to delete a
488 TagBinding.
489 retry (google.api_core.retry.Retry): Designation of what errors, if any,
490 should be retried.
491 timeout (float): The timeout for this request.
492 metadata (Sequence[Tuple[str, str]]): Strings which should be
493 sent along with the request as metadata.
495 Returns:
496 ~.operations_pb2.Operation:
497 This resource represents a
498 long-running operation that is the
499 result of a network API call.
501 """
503 http_options: List[Dict[str, str]] = [
504 {
505 "method": "delete",
506 "uri": "/v3/{name=tagBindings/**}",
507 },
508 ]
509 request, metadata = self._interceptor.pre_delete_tag_binding(
510 request, metadata
511 )
512 pb_request = tag_bindings.DeleteTagBindingRequest.pb(request)
513 transcoded_request = path_template.transcode(http_options, pb_request)
515 uri = transcoded_request["uri"]
516 method = transcoded_request["method"]
518 # Jsonify the query params
519 query_params = json.loads(
520 json_format.MessageToJson(
521 transcoded_request["query_params"],
522 including_default_value_fields=False,
523 use_integers_for_enums=True,
524 )
525 )
526 query_params.update(self._get_unset_required_fields(query_params))
528 query_params["$alt"] = "json;enum-encoding=int"
530 # Send the request
531 headers = dict(metadata)
532 headers["Content-Type"] = "application/json"
533 response = getattr(self._session, method)(
534 "{host}{uri}".format(host=self._host, uri=uri),
535 timeout=timeout,
536 headers=headers,
537 params=rest_helpers.flatten_query_params(query_params, strict=True),
538 )
540 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
541 # subclass.
542 if response.status_code >= 400:
543 raise core_exceptions.from_http_response(response)
545 # Return the response
546 resp = operations_pb2.Operation()
547 json_format.Parse(response.content, resp, ignore_unknown_fields=True)
548 resp = self._interceptor.post_delete_tag_binding(resp)
549 return resp
551 class _ListEffectiveTags(TagBindingsRestStub):
552 def __hash__(self):
553 return hash("ListEffectiveTags")
555 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
556 "parent": "",
557 }
559 @classmethod
560 def _get_unset_required_fields(cls, message_dict):
561 return {
562 k: v
563 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
564 if k not in message_dict
565 }
567 def __call__(
568 self,
569 request: tag_bindings.ListEffectiveTagsRequest,
570 *,
571 retry: OptionalRetry = gapic_v1.method.DEFAULT,
572 timeout: Optional[float] = None,
573 metadata: Sequence[Tuple[str, str]] = (),
574 ) -> tag_bindings.ListEffectiveTagsResponse:
575 r"""Call the list effective tags method over HTTP.
577 Args:
578 request (~.tag_bindings.ListEffectiveTagsRequest):
579 The request object. The request message to
580 ListEffectiveTags
581 retry (google.api_core.retry.Retry): Designation of what errors, if any,
582 should be retried.
583 timeout (float): The timeout for this request.
584 metadata (Sequence[Tuple[str, str]]): Strings which should be
585 sent along with the request as metadata.
587 Returns:
588 ~.tag_bindings.ListEffectiveTagsResponse:
589 The response of ListEffectiveTags.
590 """
592 http_options: List[Dict[str, str]] = [
593 {
594 "method": "get",
595 "uri": "/v3/effectiveTags",
596 },
597 ]
598 request, metadata = self._interceptor.pre_list_effective_tags(
599 request, metadata
600 )
601 pb_request = tag_bindings.ListEffectiveTagsRequest.pb(request)
602 transcoded_request = path_template.transcode(http_options, pb_request)
604 uri = transcoded_request["uri"]
605 method = transcoded_request["method"]
607 # Jsonify the query params
608 query_params = json.loads(
609 json_format.MessageToJson(
610 transcoded_request["query_params"],
611 including_default_value_fields=False,
612 use_integers_for_enums=True,
613 )
614 )
615 query_params.update(self._get_unset_required_fields(query_params))
617 query_params["$alt"] = "json;enum-encoding=int"
619 # Send the request
620 headers = dict(metadata)
621 headers["Content-Type"] = "application/json"
622 response = getattr(self._session, method)(
623 "{host}{uri}".format(host=self._host, uri=uri),
624 timeout=timeout,
625 headers=headers,
626 params=rest_helpers.flatten_query_params(query_params, strict=True),
627 )
629 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
630 # subclass.
631 if response.status_code >= 400:
632 raise core_exceptions.from_http_response(response)
634 # Return the response
635 resp = tag_bindings.ListEffectiveTagsResponse()
636 pb_resp = tag_bindings.ListEffectiveTagsResponse.pb(resp)
638 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
639 resp = self._interceptor.post_list_effective_tags(resp)
640 return resp
642 class _ListTagBindings(TagBindingsRestStub):
643 def __hash__(self):
644 return hash("ListTagBindings")
646 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {
647 "parent": "",
648 }
650 @classmethod
651 def _get_unset_required_fields(cls, message_dict):
652 return {
653 k: v
654 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items()
655 if k not in message_dict
656 }
658 def __call__(
659 self,
660 request: tag_bindings.ListTagBindingsRequest,
661 *,
662 retry: OptionalRetry = gapic_v1.method.DEFAULT,
663 timeout: Optional[float] = None,
664 metadata: Sequence[Tuple[str, str]] = (),
665 ) -> tag_bindings.ListTagBindingsResponse:
666 r"""Call the list tag bindings method over HTTP.
668 Args:
669 request (~.tag_bindings.ListTagBindingsRequest):
670 The request object. The request message to list all
671 TagBindings for a parent.
672 retry (google.api_core.retry.Retry): Designation of what errors, if any,
673 should be retried.
674 timeout (float): The timeout for this request.
675 metadata (Sequence[Tuple[str, str]]): Strings which should be
676 sent along with the request as metadata.
678 Returns:
679 ~.tag_bindings.ListTagBindingsResponse:
680 The ListTagBindings response.
681 """
683 http_options: List[Dict[str, str]] = [
684 {
685 "method": "get",
686 "uri": "/v3/tagBindings",
687 },
688 ]
689 request, metadata = self._interceptor.pre_list_tag_bindings(
690 request, metadata
691 )
692 pb_request = tag_bindings.ListTagBindingsRequest.pb(request)
693 transcoded_request = path_template.transcode(http_options, pb_request)
695 uri = transcoded_request["uri"]
696 method = transcoded_request["method"]
698 # Jsonify the query params
699 query_params = json.loads(
700 json_format.MessageToJson(
701 transcoded_request["query_params"],
702 including_default_value_fields=False,
703 use_integers_for_enums=True,
704 )
705 )
706 query_params.update(self._get_unset_required_fields(query_params))
708 query_params["$alt"] = "json;enum-encoding=int"
710 # Send the request
711 headers = dict(metadata)
712 headers["Content-Type"] = "application/json"
713 response = getattr(self._session, method)(
714 "{host}{uri}".format(host=self._host, uri=uri),
715 timeout=timeout,
716 headers=headers,
717 params=rest_helpers.flatten_query_params(query_params, strict=True),
718 )
720 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
721 # subclass.
722 if response.status_code >= 400:
723 raise core_exceptions.from_http_response(response)
725 # Return the response
726 resp = tag_bindings.ListTagBindingsResponse()
727 pb_resp = tag_bindings.ListTagBindingsResponse.pb(resp)
729 json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
730 resp = self._interceptor.post_list_tag_bindings(resp)
731 return resp
733 @property
734 def create_tag_binding(
735 self,
736 ) -> Callable[[tag_bindings.CreateTagBindingRequest], operations_pb2.Operation]:
737 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
738 # In C++ this would require a dynamic_cast
739 return self._CreateTagBinding(self._session, self._host, self._interceptor) # type: ignore
741 @property
742 def delete_tag_binding(
743 self,
744 ) -> Callable[[tag_bindings.DeleteTagBindingRequest], operations_pb2.Operation]:
745 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
746 # In C++ this would require a dynamic_cast
747 return self._DeleteTagBinding(self._session, self._host, self._interceptor) # type: ignore
749 @property
750 def list_effective_tags(
751 self,
752 ) -> Callable[
753 [tag_bindings.ListEffectiveTagsRequest], tag_bindings.ListEffectiveTagsResponse
754 ]:
755 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
756 # In C++ this would require a dynamic_cast
757 return self._ListEffectiveTags(self._session, self._host, self._interceptor) # type: ignore
759 @property
760 def list_tag_bindings(
761 self,
762 ) -> Callable[
763 [tag_bindings.ListTagBindingsRequest], tag_bindings.ListTagBindingsResponse
764 ]:
765 # The return type is fine, but mypy isn't sophisticated enough to determine what's going on here.
766 # In C++ this would require a dynamic_cast
767 return self._ListTagBindings(self._session, self._host, self._interceptor) # type: ignore
769 @property
770 def get_operation(self):
771 return self._GetOperation(self._session, self._host, self._interceptor) # type: ignore
773 class _GetOperation(TagBindingsRestStub):
774 def __call__(
775 self,
776 request: operations_pb2.GetOperationRequest,
777 *,
778 retry: OptionalRetry = gapic_v1.method.DEFAULT,
779 timeout: Optional[float] = None,
780 metadata: Sequence[Tuple[str, str]] = (),
781 ) -> operations_pb2.Operation:
783 r"""Call the get operation method over HTTP.
785 Args:
786 request (operations_pb2.GetOperationRequest):
787 The request object for GetOperation method.
788 retry (google.api_core.retry.Retry): Designation of what errors, if any,
789 should be retried.
790 timeout (float): The timeout for this request.
791 metadata (Sequence[Tuple[str, str]]): Strings which should be
792 sent along with the request as metadata.
794 Returns:
795 operations_pb2.Operation: Response from GetOperation method.
796 """
798 http_options: List[Dict[str, str]] = [
799 {
800 "method": "get",
801 "uri": "/v3/{name=operations/**}",
802 },
803 ]
805 request, metadata = self._interceptor.pre_get_operation(request, metadata)
806 request_kwargs = json_format.MessageToDict(request)
807 transcoded_request = path_template.transcode(http_options, **request_kwargs)
809 uri = transcoded_request["uri"]
810 method = transcoded_request["method"]
812 # Jsonify the query params
813 query_params = json.loads(json.dumps(transcoded_request["query_params"]))
815 # Send the request
816 headers = dict(metadata)
817 headers["Content-Type"] = "application/json"
819 response = getattr(self._session, method)(
820 "{host}{uri}".format(host=self._host, uri=uri),
821 timeout=timeout,
822 headers=headers,
823 params=rest_helpers.flatten_query_params(query_params),
824 )
826 # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
827 # subclass.
828 if response.status_code >= 400:
829 raise core_exceptions.from_http_response(response)
831 resp = operations_pb2.Operation()
832 resp = json_format.Parse(response.content.decode("utf-8"), resp)
833 resp = self._interceptor.post_get_operation(resp)
834 return resp
836 @property
837 def kind(self) -> str:
838 return "rest"
840 def close(self):
841 self._session.close()
844__all__ = ("TagBindingsRestTransport",)