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 inspect
17import json
18import logging as std_logging
19import pickle
20from typing import Awaitable, Callable, Dict, Optional, Sequence, Tuple, Union
21import warnings
23from google.api_core import exceptions as core_exceptions
24from google.api_core import gapic_v1, grpc_helpers_async
25from google.api_core import retry_async as retries
26from google.auth import credentials as ga_credentials # type: ignore
27from google.auth.transport.grpc import SslCredentials # type: ignore
28from google.protobuf.json_format import MessageToJson
29import google.protobuf.message
30import grpc # type: ignore
31from grpc.experimental import aio # type: ignore
32import proto # type: ignore
34from google.cloud.iam_credentials_v1.types import common
36from .base import DEFAULT_CLIENT_INFO, IAMCredentialsTransport
37from .grpc import IAMCredentialsGrpcTransport
39try:
40 from google.api_core import client_logging # type: ignore
42 CLIENT_LOGGING_SUPPORTED = True # pragma: NO COVER
43except ImportError: # pragma: NO COVER
44 CLIENT_LOGGING_SUPPORTED = False
46_LOGGER = std_logging.getLogger(__name__)
49class _LoggingClientAIOInterceptor(
50 grpc.aio.UnaryUnaryClientInterceptor
51): # pragma: NO COVER
52 async def intercept_unary_unary(self, continuation, client_call_details, request):
53 logging_enabled = CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(
54 std_logging.DEBUG
55 )
56 if logging_enabled: # pragma: NO COVER
57 request_metadata = client_call_details.metadata
58 if isinstance(request, proto.Message):
59 request_payload = type(request).to_json(request)
60 elif isinstance(request, google.protobuf.message.Message):
61 request_payload = MessageToJson(request)
62 else:
63 request_payload = f"{type(request).__name__}: {pickle.dumps(request)}"
65 request_metadata = {
66 key: value.decode("utf-8") if isinstance(value, bytes) else value
67 for key, value in request_metadata
68 }
69 grpc_request = {
70 "payload": request_payload,
71 "requestMethod": "grpc",
72 "metadata": dict(request_metadata),
73 }
74 _LOGGER.debug(
75 f"Sending request for {client_call_details.method}",
76 extra={
77 "serviceName": "google.iam.credentials.v1.IAMCredentials",
78 "rpcName": str(client_call_details.method),
79 "request": grpc_request,
80 "metadata": grpc_request["metadata"],
81 },
82 )
83 response = await continuation(client_call_details, request)
84 if logging_enabled: # pragma: NO COVER
85 response_metadata = await response.trailing_metadata()
86 # Convert gRPC metadata `<class 'grpc.aio._metadata.Metadata'>` to list of tuples
87 metadata = (
88 dict([(k, str(v)) for k, v in response_metadata])
89 if response_metadata
90 else None
91 )
92 result = await response
93 if isinstance(result, proto.Message):
94 response_payload = type(result).to_json(result)
95 elif isinstance(result, google.protobuf.message.Message):
96 response_payload = MessageToJson(result)
97 else:
98 response_payload = f"{type(result).__name__}: {pickle.dumps(result)}"
99 grpc_response = {
100 "payload": response_payload,
101 "metadata": metadata,
102 "status": "OK",
103 }
104 _LOGGER.debug(
105 f"Received response to rpc {client_call_details.method}.",
106 extra={
107 "serviceName": "google.iam.credentials.v1.IAMCredentials",
108 "rpcName": str(client_call_details.method),
109 "response": grpc_response,
110 "metadata": grpc_response["metadata"],
111 },
112 )
113 return response
116class IAMCredentialsGrpcAsyncIOTransport(IAMCredentialsTransport):
117 """gRPC AsyncIO backend transport for IAMCredentials.
119 A service account is a special type of Google account that
120 belongs to your application or a virtual machine (VM), instead
121 of to an individual end user. Your application assumes the
122 identity of the service account to call Google APIs, so that the
123 users aren't directly involved.
125 Service account credentials are used to temporarily assume the
126 identity of the service account. Supported credential types
127 include OAuth 2.0 access tokens, OpenID Connect ID tokens,
128 self-signed JSON Web Tokens (JWTs), and more.
130 This class defines the same methods as the primary client, so the
131 primary client can load the underlying transport implementation
132 and call it.
134 It sends protocol buffers over the wire using gRPC (which is built on
135 top of HTTP/2); the ``grpcio`` package must be installed.
136 """
138 _grpc_channel: aio.Channel
139 _stubs: Dict[str, Callable] = {}
141 @classmethod
142 def create_channel(
143 cls,
144 host: str = "iamcredentials.googleapis.com",
145 credentials: Optional[ga_credentials.Credentials] = None,
146 credentials_file: Optional[str] = None,
147 scopes: Optional[Sequence[str]] = None,
148 quota_project_id: Optional[str] = None,
149 **kwargs,
150 ) -> aio.Channel:
151 """Create and return a gRPC AsyncIO channel object.
152 Args:
153 host (Optional[str]): The host for the channel to use.
154 credentials (Optional[~.Credentials]): The
155 authorization credentials to attach to requests. These
156 credentials identify this application to the service. If
157 none are specified, the client will attempt to ascertain
158 the credentials from the environment.
159 credentials_file (Optional[str]): A file with credentials that can
160 be loaded with :func:`google.auth.load_credentials_from_file`.
161 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
162 service. These are only used when credentials are not specified and
163 are passed to :func:`google.auth.default`.
164 quota_project_id (Optional[str]): An optional project to use for billing
165 and quota.
166 kwargs (Optional[dict]): Keyword arguments, which are passed to the
167 channel creation.
168 Returns:
169 aio.Channel: A gRPC AsyncIO channel object.
170 """
172 return grpc_helpers_async.create_channel(
173 host,
174 credentials=credentials,
175 credentials_file=credentials_file,
176 quota_project_id=quota_project_id,
177 default_scopes=cls.AUTH_SCOPES,
178 scopes=scopes,
179 default_host=cls.DEFAULT_HOST,
180 **kwargs,
181 )
183 def __init__(
184 self,
185 *,
186 host: str = "iamcredentials.googleapis.com",
187 credentials: Optional[ga_credentials.Credentials] = None,
188 credentials_file: Optional[str] = None,
189 scopes: Optional[Sequence[str]] = None,
190 channel: Optional[Union[aio.Channel, Callable[..., aio.Channel]]] = None,
191 api_mtls_endpoint: Optional[str] = None,
192 client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
193 ssl_channel_credentials: Optional[grpc.ChannelCredentials] = None,
194 client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
195 quota_project_id: Optional[str] = None,
196 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
197 always_use_jwt_access: Optional[bool] = False,
198 api_audience: Optional[str] = None,
199 ) -> None:
200 """Instantiate the transport.
202 Args:
203 host (Optional[str]):
204 The hostname to connect to (default: 'iamcredentials.googleapis.com').
205 credentials (Optional[google.auth.credentials.Credentials]): The
206 authorization credentials to attach to requests. These
207 credentials identify the application to the service; if none
208 are specified, the client will attempt to ascertain the
209 credentials from the environment.
210 This argument is ignored if a ``channel`` instance is provided.
211 credentials_file (Optional[str]): A file with credentials that can
212 be loaded with :func:`google.auth.load_credentials_from_file`.
213 This argument is ignored if a ``channel`` instance is provided.
214 scopes (Optional[Sequence[str]]): A optional list of scopes needed for this
215 service. These are only used when credentials are not specified and
216 are passed to :func:`google.auth.default`.
217 channel (Optional[Union[aio.Channel, Callable[..., aio.Channel]]]):
218 A ``Channel`` instance through which to make calls, or a Callable
219 that constructs and returns one. If set to None, ``self.create_channel``
220 is used to create the channel. If a Callable is given, it will be called
221 with the same arguments as used in ``self.create_channel``.
222 api_mtls_endpoint (Optional[str]): Deprecated. The mutual TLS endpoint.
223 If provided, it overrides the ``host`` argument and tries to create
224 a mutual TLS channel with client SSL credentials from
225 ``client_cert_source`` or application default SSL credentials.
226 client_cert_source (Optional[Callable[[], Tuple[bytes, bytes]]]):
227 Deprecated. A callback to provide client SSL certificate bytes and
228 private key bytes, both in PEM format. It is ignored if
229 ``api_mtls_endpoint`` is None.
230 ssl_channel_credentials (grpc.ChannelCredentials): SSL credentials
231 for the grpc channel. It is ignored if a ``channel`` instance is provided.
232 client_cert_source_for_mtls (Optional[Callable[[], Tuple[bytes, bytes]]]):
233 A callback to provide client certificate bytes and private key bytes,
234 both in PEM format. It is used to configure a mutual TLS channel. It is
235 ignored if a ``channel`` instance or ``ssl_channel_credentials`` is provided.
236 quota_project_id (Optional[str]): An optional project to use for billing
237 and quota.
238 client_info (google.api_core.gapic_v1.client_info.ClientInfo):
239 The client info used to send a user-agent string along with
240 API requests. If ``None``, then default info will be used.
241 Generally, you only need to set this if you're developing
242 your own client library.
243 always_use_jwt_access (Optional[bool]): Whether self signed JWT should
244 be used for service account credentials.
246 Raises:
247 google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
248 creation failed for any reason.
249 google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
250 and ``credentials_file`` are passed.
251 """
252 self._grpc_channel = None
253 self._ssl_channel_credentials = ssl_channel_credentials
254 self._stubs: Dict[str, Callable] = {}
256 if api_mtls_endpoint:
257 warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
258 if client_cert_source:
259 warnings.warn("client_cert_source is deprecated", DeprecationWarning)
261 if isinstance(channel, aio.Channel):
262 # Ignore credentials if a channel was passed.
263 credentials = None
264 self._ignore_credentials = True
265 # If a channel was explicitly provided, set it.
266 self._grpc_channel = channel
267 self._ssl_channel_credentials = None
268 else:
269 if api_mtls_endpoint:
270 host = api_mtls_endpoint
272 # Create SSL credentials with client_cert_source or application
273 # default SSL credentials.
274 if client_cert_source:
275 cert, key = client_cert_source()
276 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
277 certificate_chain=cert, private_key=key
278 )
279 else:
280 self._ssl_channel_credentials = SslCredentials().ssl_credentials
282 else:
283 if client_cert_source_for_mtls and not ssl_channel_credentials:
284 cert, key = client_cert_source_for_mtls()
285 self._ssl_channel_credentials = grpc.ssl_channel_credentials(
286 certificate_chain=cert, private_key=key
287 )
289 # The base transport sets the host, credentials and scopes
290 super().__init__(
291 host=host,
292 credentials=credentials,
293 credentials_file=credentials_file,
294 scopes=scopes,
295 quota_project_id=quota_project_id,
296 client_info=client_info,
297 always_use_jwt_access=always_use_jwt_access,
298 api_audience=api_audience,
299 )
301 if not self._grpc_channel:
302 # initialize with the provided callable or the default channel
303 channel_init = channel or type(self).create_channel
304 self._grpc_channel = channel_init(
305 self._host,
306 # use the credentials which are saved
307 credentials=self._credentials,
308 # Set ``credentials_file`` to ``None`` here as
309 # the credentials that we saved earlier should be used.
310 credentials_file=None,
311 scopes=self._scopes,
312 ssl_credentials=self._ssl_channel_credentials,
313 quota_project_id=quota_project_id,
314 options=[
315 ("grpc.max_send_message_length", -1),
316 ("grpc.max_receive_message_length", -1),
317 ],
318 )
320 self._interceptor = _LoggingClientAIOInterceptor()
321 self._grpc_channel._unary_unary_interceptors.append(self._interceptor)
322 self._logged_channel = self._grpc_channel
323 self._wrap_with_kind = (
324 "kind" in inspect.signature(gapic_v1.method_async.wrap_method).parameters
325 )
326 # Wrap messages. This must be done after self._logged_channel exists
327 self._prep_wrapped_messages(client_info)
329 @property
330 def grpc_channel(self) -> aio.Channel:
331 """Create the channel designed to connect to this service.
333 This property caches on the instance; repeated calls return
334 the same channel.
335 """
336 # Return the channel from cache.
337 return self._grpc_channel
339 @property
340 def generate_access_token(
341 self,
342 ) -> Callable[
343 [common.GenerateAccessTokenRequest],
344 Awaitable[common.GenerateAccessTokenResponse],
345 ]:
346 r"""Return a callable for the generate access token method over gRPC.
348 Generates an OAuth 2.0 access token for a service
349 account.
351 Returns:
352 Callable[[~.GenerateAccessTokenRequest],
353 Awaitable[~.GenerateAccessTokenResponse]]:
354 A function that, when called, will call the underlying RPC
355 on the server.
356 """
357 # Generate a "stub function" on-the-fly which will actually make
358 # the request.
359 # gRPC handles serialization and deserialization, so we just need
360 # to pass in the functions for each.
361 if "generate_access_token" not in self._stubs:
362 self._stubs["generate_access_token"] = self._logged_channel.unary_unary(
363 "/google.iam.credentials.v1.IAMCredentials/GenerateAccessToken",
364 request_serializer=common.GenerateAccessTokenRequest.serialize,
365 response_deserializer=common.GenerateAccessTokenResponse.deserialize,
366 )
367 return self._stubs["generate_access_token"]
369 @property
370 def generate_id_token(
371 self,
372 ) -> Callable[
373 [common.GenerateIdTokenRequest], Awaitable[common.GenerateIdTokenResponse]
374 ]:
375 r"""Return a callable for the generate id token method over gRPC.
377 Generates an OpenID Connect ID token for a service
378 account.
380 Returns:
381 Callable[[~.GenerateIdTokenRequest],
382 Awaitable[~.GenerateIdTokenResponse]]:
383 A function that, when called, will call the underlying RPC
384 on the server.
385 """
386 # Generate a "stub function" on-the-fly which will actually make
387 # the request.
388 # gRPC handles serialization and deserialization, so we just need
389 # to pass in the functions for each.
390 if "generate_id_token" not in self._stubs:
391 self._stubs["generate_id_token"] = self._logged_channel.unary_unary(
392 "/google.iam.credentials.v1.IAMCredentials/GenerateIdToken",
393 request_serializer=common.GenerateIdTokenRequest.serialize,
394 response_deserializer=common.GenerateIdTokenResponse.deserialize,
395 )
396 return self._stubs["generate_id_token"]
398 @property
399 def sign_blob(
400 self,
401 ) -> Callable[[common.SignBlobRequest], Awaitable[common.SignBlobResponse]]:
402 r"""Return a callable for the sign blob method over gRPC.
404 Signs a blob using a service account's system-managed
405 private key.
407 Returns:
408 Callable[[~.SignBlobRequest],
409 Awaitable[~.SignBlobResponse]]:
410 A function that, when called, will call the underlying RPC
411 on the server.
412 """
413 # Generate a "stub function" on-the-fly which will actually make
414 # the request.
415 # gRPC handles serialization and deserialization, so we just need
416 # to pass in the functions for each.
417 if "sign_blob" not in self._stubs:
418 self._stubs["sign_blob"] = self._logged_channel.unary_unary(
419 "/google.iam.credentials.v1.IAMCredentials/SignBlob",
420 request_serializer=common.SignBlobRequest.serialize,
421 response_deserializer=common.SignBlobResponse.deserialize,
422 )
423 return self._stubs["sign_blob"]
425 @property
426 def sign_jwt(
427 self,
428 ) -> Callable[[common.SignJwtRequest], Awaitable[common.SignJwtResponse]]:
429 r"""Return a callable for the sign jwt method over gRPC.
431 Signs a JWT using a service account's system-managed
432 private key.
434 Returns:
435 Callable[[~.SignJwtRequest],
436 Awaitable[~.SignJwtResponse]]:
437 A function that, when called, will call the underlying RPC
438 on the server.
439 """
440 # Generate a "stub function" on-the-fly which will actually make
441 # the request.
442 # gRPC handles serialization and deserialization, so we just need
443 # to pass in the functions for each.
444 if "sign_jwt" not in self._stubs:
445 self._stubs["sign_jwt"] = self._logged_channel.unary_unary(
446 "/google.iam.credentials.v1.IAMCredentials/SignJwt",
447 request_serializer=common.SignJwtRequest.serialize,
448 response_deserializer=common.SignJwtResponse.deserialize,
449 )
450 return self._stubs["sign_jwt"]
452 def _prep_wrapped_messages(self, client_info):
453 """Precompute the wrapped methods, overriding the base class method to use async wrappers."""
454 self._wrapped_methods = {
455 self.generate_access_token: self._wrap_method(
456 self.generate_access_token,
457 default_retry=retries.AsyncRetry(
458 initial=0.1,
459 maximum=60.0,
460 multiplier=1.3,
461 predicate=retries.if_exception_type(
462 core_exceptions.DeadlineExceeded,
463 core_exceptions.ServiceUnavailable,
464 ),
465 deadline=60.0,
466 ),
467 default_timeout=60.0,
468 client_info=client_info,
469 ),
470 self.generate_id_token: self._wrap_method(
471 self.generate_id_token,
472 default_retry=retries.AsyncRetry(
473 initial=0.1,
474 maximum=60.0,
475 multiplier=1.3,
476 predicate=retries.if_exception_type(
477 core_exceptions.DeadlineExceeded,
478 core_exceptions.ServiceUnavailable,
479 ),
480 deadline=60.0,
481 ),
482 default_timeout=60.0,
483 client_info=client_info,
484 ),
485 self.sign_blob: self._wrap_method(
486 self.sign_blob,
487 default_retry=retries.AsyncRetry(
488 initial=0.1,
489 maximum=60.0,
490 multiplier=1.3,
491 predicate=retries.if_exception_type(
492 core_exceptions.DeadlineExceeded,
493 core_exceptions.ServiceUnavailable,
494 ),
495 deadline=60.0,
496 ),
497 default_timeout=60.0,
498 client_info=client_info,
499 ),
500 self.sign_jwt: self._wrap_method(
501 self.sign_jwt,
502 default_retry=retries.AsyncRetry(
503 initial=0.1,
504 maximum=60.0,
505 multiplier=1.3,
506 predicate=retries.if_exception_type(
507 core_exceptions.DeadlineExceeded,
508 core_exceptions.ServiceUnavailable,
509 ),
510 deadline=60.0,
511 ),
512 default_timeout=60.0,
513 client_info=client_info,
514 ),
515 }
517 def _wrap_method(self, func, *args, **kwargs):
518 if self._wrap_with_kind: # pragma: NO COVER
519 kwargs["kind"] = self.kind
520 return gapic_v1.method_async.wrap_method(func, *args, **kwargs)
522 def close(self):
523 return self._logged_channel.close()
525 @property
526 def kind(self) -> str:
527 return "grpc_asyncio"
530__all__ = ("IAMCredentialsGrpcAsyncIOTransport",)