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# Copyright 2017 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Helpers for :mod:`grpc`."""
17import collections
18import functools
19import os
20import urllib.parse
21import warnings
22from typing import (
23 Callable,
24 Generic,
25 Iterator,
26 Optional,
27 Sequence,
28 TypeAlias,
29 TypeVar,
30 cast,
31 get_args,
32)
34import google.auth
35import google.auth.credentials
36import google.auth.transport.grpc
37import google.auth.transport.requests
38import google.protobuf
39import grpc
41from google.api_core import exceptions, general_helpers
42from google.api_core._feature_gating_helpers import _strtobool
44_DIRECT_PATH_INTERCONNECT_ENV = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT"
47def _resolve_direct_path_interconnect(
48 attempt_direct_path_xds_over_interconnect: Optional[bool] = None,
49) -> bool:
50 """Resolves whether DirectPath over Interconnect is enabled.
52 Args:
53 attempt_direct_path_xds_over_interconnect (Optional[bool]): Programmatic
54 override. If specified, this value takes precedence over environment
55 variables. If None, falls back to the environment variable.
57 Returns:
58 bool: True if DirectPath over Interconnect is enabled, False otherwise.
60 Raises:
61 ValueError: If the environment variable contains an invalid boolean value.
62 """
63 if attempt_direct_path_xds_over_interconnect is not None:
64 return attempt_direct_path_xds_over_interconnect
66 env_val = os.environ.get(_DIRECT_PATH_INTERCONNECT_ENV)
67 if env_val is not None:
68 try:
69 val = _strtobool(env_val)
70 except ValueError:
71 val = None
72 if val is None:
73 raise ValueError(
74 f"Invalid value for {_DIRECT_PATH_INTERCONNECT_ENV}: {env_val!r}"
75 )
76 return val
78 return False
81def _extract_target_host(target: str) -> str:
82 """Extracts the host from a target URI or address."""
83 stripped_target = target
84 for prefix in ("google-c2p:///", "dns:///", "https://", "http://"):
85 if target.startswith(prefix):
86 stripped_target = target.removeprefix(prefix)
87 break
88 try:
89 return urllib.parse.urlsplit(f"//{stripped_target}").hostname or target
90 except ValueError:
91 return target
94def _extract_direct_path_authority(target: str) -> Optional[str]:
95 """Extracts the canonical TLS/HTTP2 authority for a ``-direct.googleapis.com`` target."""
96 clean_host = _extract_target_host(target)
97 suffix = "-direct.googleapis.com"
98 if clean_host.endswith(suffix):
99 service_prefix = clean_host.removesuffix(suffix)
100 if service_prefix:
101 return f"{service_prefix}.googleapis.com"
102 return None
105# The list of gRPC Callable interfaces that return iterators.
106_STREAM_WRAP_CLASSES = (grpc.UnaryStreamMultiCallable, grpc.StreamStreamMultiCallable)
108# denotes the proto response type for grpc calls
109P = TypeVar("P")
111# Type alias representing any client-side gRPC interceptor
112ClientInterceptor: TypeAlias = (
113 grpc.UnaryUnaryClientInterceptor
114 | grpc.UnaryStreamClientInterceptor
115 | grpc.StreamUnaryClientInterceptor
116 | grpc.StreamStreamClientInterceptor
117)
119# Runtime tuple of gRPC client interceptor base classes for isinstance checks
120_CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
123def _patch_callable_name(callable_):
124 """Fix-up gRPC callable attributes.
126 gRPC callable lack the ``__name__`` attribute which causes
127 :func:`functools.wraps` to error. This adds the attribute if needed.
128 """
129 if not hasattr(callable_, "__name__"):
130 callable_.__name__ = callable_.__class__.__name__
133def _wrap_unary_errors(callable_):
134 """Map errors for Unary-Unary and Stream-Unary gRPC callables."""
135 _patch_callable_name(callable_)
137 @functools.wraps(callable_)
138 def error_remapped_callable(*args, **kwargs):
139 try:
140 return callable_(*args, **kwargs)
141 except grpc.RpcError as exc:
142 raise exceptions.from_grpc_error(exc) from exc
144 return error_remapped_callable
147class _StreamingResponseIterator(Generic[P], grpc.Call):
148 def __init__(self, wrapped, prefetch_first_result=True):
149 self._wrapped = wrapped
151 # This iterator is used in a retry context, and returned outside after init.
152 # gRPC will not throw an exception until the stream is consumed, so we need
153 # to retrieve the first result, in order to fail, in order to trigger a retry.
154 try:
155 if prefetch_first_result:
156 self._stored_first_result = next(self._wrapped)
157 except TypeError:
158 # It is possible the wrapped method isn't an iterable (a grpc.Call
159 # for instance). If this happens don't store the first result.
160 pass
161 except StopIteration:
162 # ignore stop iteration at this time. This should be handled outside of retry.
163 pass
165 def __iter__(self) -> Iterator[P]:
166 """This iterator is also an iterable that returns itself."""
167 return self
169 def __next__(self) -> P:
170 """Get the next response from the stream.
172 Returns:
173 protobuf.Message: A single response from the stream.
174 """
175 try:
176 if hasattr(self, "_stored_first_result"):
177 result = self._stored_first_result
178 del self._stored_first_result
179 return result
180 return next(self._wrapped)
181 except grpc.RpcError as exc:
182 # If the stream has already returned data, we cannot recover here.
183 raise exceptions.from_grpc_error(exc) from exc
185 # grpc.Call & grpc.RpcContext interface
187 def add_callback(self, callback):
188 return self._wrapped.add_callback(callback)
190 def cancel(self):
191 return self._wrapped.cancel()
193 def code(self):
194 return self._wrapped.code()
196 def details(self):
197 return self._wrapped.details()
199 def initial_metadata(self):
200 return self._wrapped.initial_metadata()
202 def is_active(self):
203 return self._wrapped.is_active()
205 def time_remaining(self):
206 return self._wrapped.time_remaining()
208 def trailing_metadata(self):
209 return self._wrapped.trailing_metadata()
212# public type alias denoting the return type of streaming gapic calls
213GrpcStream = _StreamingResponseIterator[P]
216def _wrap_stream_errors(callable_):
217 """Wrap errors for Unary-Stream and Stream-Stream gRPC callables.
219 The callables that return iterators require a bit more logic to re-map
220 errors when iterating. This wraps both the initial invocation and the
221 iterator of the return value to re-map errors.
222 """
223 _patch_callable_name(callable_)
225 @functools.wraps(callable_)
226 def error_remapped_callable(*args, **kwargs):
227 try:
228 result = callable_(*args, **kwargs)
229 # Auto-fetching the first result causes PubSub client's streaming pull
230 # to hang when re-opening the stream, thus we need examine the hacky
231 # hidden flag to see if pre-fetching is disabled.
232 # https://github.com/googleapis/python-pubsub/issues/93#issuecomment-630762257
233 prefetch_first = getattr(callable_, "_prefetch_first_result_", True)
234 return _StreamingResponseIterator(
235 result, prefetch_first_result=prefetch_first
236 )
237 except grpc.RpcError as exc:
238 raise exceptions.from_grpc_error(exc) from exc
240 return error_remapped_callable
243def wrap_errors(callable_):
244 """Wrap a gRPC callable and map :class:`grpc.RpcErrors` to friendly error
245 classes.
247 Errors raised by the gRPC callable are mapped to the appropriate
248 :class:`google.api_core.exceptions.GoogleAPICallError` subclasses.
249 The original `grpc.RpcError` (which is usually also a `grpc.Call`) is
250 available from the ``response`` property on the mapped exception. This
251 is useful for extracting metadata from the original error.
253 Args:
254 callable_ (Callable): A gRPC callable.
256 Returns:
257 Callable: The wrapped gRPC callable.
258 """
259 if isinstance(callable_, _STREAM_WRAP_CLASSES):
260 return _wrap_stream_errors(callable_)
261 else:
262 return _wrap_unary_errors(callable_)
265def _create_composite_credentials(
266 credentials=None,
267 credentials_file=None,
268 default_scopes=None,
269 scopes=None,
270 ssl_credentials=None,
271 quota_project_id=None,
272 default_host=None,
273):
274 """Create the composite credentials for secure channels.
276 Args:
277 credentials (google.auth.credentials.Credentials): The credentials. If
278 not specified, then this function will attempt to ascertain the
279 credentials from the environment using :func:`google.auth.default`.
280 credentials_file (str): Deprecated. A file with credentials that can be loaded with
281 :func:`google.auth.load_credentials_from_file`. This argument is
282 mutually exclusive with credentials. This argument will be
283 removed in the next major version of `google-api-core`.
285 .. warning::
286 Important: If you accept a credential configuration (credential JSON/File/Stream)
287 from an external source for authentication to Google Cloud Platform, you must
288 validate it before providing it to any Google API or client library. Providing an
289 unvalidated credential configuration to Google APIs or libraries can compromise
290 the security of your systems and data. For more information, refer to
291 `Validate credential configurations from external sources`_.
293 .. _Validate credential configurations from external sources:
295 https://cloud.google.com/docs/authentication/external/externally-sourced-credentials
296 default_scopes (Sequence[str]): A optional list of scopes needed for this
297 service. These are only used when credentials are not specified and
298 are passed to :func:`google.auth.default`.
299 scopes (Sequence[str]): A optional list of scopes needed for this
300 service. These are only used when credentials are not specified and
301 are passed to :func:`google.auth.default`.
302 ssl_credentials (grpc.ChannelCredentials): Optional SSL channel
303 credentials. This can be used to specify different certificates.
304 quota_project_id (str): An optional project to use for billing and quota.
305 default_host (str): The default endpoint. e.g., "pubsub.googleapis.com".
307 Returns:
308 grpc.ChannelCredentials: The composed channel credentials object.
310 Raises:
311 google.api_core.DuplicateCredentialArgs: If both a credentials object and credentials_file are passed.
312 """
313 if credentials_file is not None:
314 warnings.warn(general_helpers._CREDENTIALS_FILE_WARNING, DeprecationWarning)
316 if credentials and credentials_file:
317 raise exceptions.DuplicateCredentialArgs(
318 "'credentials' and 'credentials_file' are mutually exclusive."
319 )
321 if credentials_file:
322 credentials, _ = google.auth.load_credentials_from_file(
323 credentials_file, scopes=scopes, default_scopes=default_scopes
324 )
325 elif credentials:
326 credentials = google.auth.credentials.with_scopes_if_required(
327 credentials, scopes=scopes, default_scopes=default_scopes
328 )
329 else:
330 credentials, _ = google.auth.default(
331 scopes=scopes, default_scopes=default_scopes
332 )
334 if quota_project_id and isinstance(
335 credentials, google.auth.credentials.CredentialsWithQuotaProject
336 ):
337 credentials = credentials.with_quota_project(quota_project_id)
339 request = google.auth.transport.requests.Request()
341 # Create the metadata plugin for inserting the authorization header.
342 try:
343 metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
344 credentials,
345 request,
346 default_host=default_host,
347 suppress_metrics_header=True,
348 )
349 except TypeError:
350 # Support older versions of google-auth that do not accept suppress_metrics_header
351 metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
352 credentials,
353 request,
354 default_host=default_host,
355 )
357 # Create a set of grpc.CallCredentials using the metadata plugin.
358 google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
360 # if `ssl_credentials` is set, use `grpc.composite_channel_credentials` instead of
361 # `grpc.compute_engine_channel_credentials` as the former supports passing
362 # `ssl_credentials` via `channel_credentials` which is needed for mTLS.
363 if ssl_credentials:
364 # Combine the ssl credentials and the authorization credentials.
365 # See https://grpc.github.io/grpc/python/grpc.html#grpc.composite_channel_credentials
366 return grpc.composite_channel_credentials(
367 ssl_credentials, google_auth_credentials
368 )
369 else:
370 # Use grpc.compute_engine_channel_credentials in order to support Direct Path.
371 # See https://grpc.github.io/grpc/python/grpc.html#grpc.compute_engine_channel_credentials
372 # TODO(https://github.com/googleapis/python-api-core/issues/598):
373 # Although `grpc.compute_engine_channel_credentials` returns channel credentials
374 # outside of a Google Compute Engine environment (GCE), we should determine if
375 # there is a way to reliably detect a GCE environment so that
376 # `grpc.compute_engine_channel_credentials` is not called outside of GCE.
377 return grpc.compute_engine_channel_credentials(google_auth_credentials)
380def _setup_direct_path_and_credentials(
381 target,
382 credentials=None,
383 scopes=None,
384 ssl_credentials=None,
385 credentials_file=None,
386 quota_project_id=None,
387 default_scopes=None,
388 default_host=None,
389 attempt_direct_path: Optional[bool] = False,
390 attempt_direct_path_xds_over_interconnect: Optional[bool] = None,
391 **kwargs,
392):
393 """Configures credentials, options, and target for DirectPath and Interconnect.
395 Args:
396 target (str): The target service address in the format 'hostname:port'.
397 credentials (google.auth.credentials.Credentials): The credentials. If
398 not specified, then this function will attempt to ascertain the
399 credentials from the environment using :func:`google.auth.default`.
400 scopes (Sequence[str]): A optional list of scopes needed for this
401 service. These are only used when credentials are not specified and
402 are passed to :func:`google.auth.default`.
403 ssl_credentials (grpc.ChannelCredentials): Optional SSL channel
404 credentials. This can be used to specify different certificates.
405 credentials_file (str): Deprecated. A file with credentials that can be
406 loaded with :func:`google.auth.load_credentials_from_file`.
407 quota_project_id (str): An optional project to use for billing and quota.
408 default_scopes (Sequence[str]): Default scopes passed by a Google client
409 library. Use 'scopes' for user-defined scopes.
410 default_host (str): The default endpoint. e.g., "pubsub.googleapis.com".
411 attempt_direct_path (Optional[bool]): If set, Direct Path will be attempted.
412 attempt_direct_path_xds_over_interconnect (Optional[bool]): If set,
413 DirectPath over Cloud Interconnect will be attempted using standard
414 TLS credentials and ``?force-xds`` C2P target resolution. If None
415 (default), the setting is resolved from the
416 ``GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT``
417 environment variable.
418 kwargs: Additional key-word args passed to the channel creation function.
420 Returns:
421 tuple[str, grpc.ChannelCredentials, dict]: A tuple containing the resolved
422 target, composite credentials, and kwargs.
424 Raises:
425 google.api_core.DuplicateCredentialArgs: If both a credentials object
426 and credentials_file are passed.
427 ValueError: If `ssl_credentials` is set and `attempt_direct_path` is
428 set to `True` without `attempt_direct_path_xds_over_interconnect`.
429 """
430 if attempt_direct_path or attempt_direct_path_xds_over_interconnect is not None:
431 use_dp_interconnect = _resolve_direct_path_interconnect(
432 attempt_direct_path_xds_over_interconnect
433 )
434 else:
435 use_dp_interconnect = False
437 # If `ssl_credentials` is set and `attempt_direct_path` is set to `True`,
438 # raise ValueError as this is not yet supported for GCE ALTS DirectPath.
439 # See https://github.com/googleapis/python-api-core/issues/590
440 if ssl_credentials and attempt_direct_path and not use_dp_interconnect:
441 raise ValueError("Using ssl_credentials with Direct Path is not supported")
443 if use_dp_interconnect and ssl_credentials is None:
444 ssl_credentials = grpc.ssl_channel_credentials()
446 composite_credentials = _create_composite_credentials(
447 credentials=credentials,
448 credentials_file=credentials_file,
449 default_scopes=default_scopes,
450 scopes=scopes,
451 ssl_credentials=ssl_credentials,
452 quota_project_id=quota_project_id,
453 default_host=default_host,
454 )
456 if use_dp_interconnect:
457 authority = _extract_direct_path_authority(target)
458 if authority:
459 existing_options = tuple(kwargs.get("options") or ())
460 option_keys = {opt[0] for opt in existing_options}
461 if (
462 "grpc.ssl_target_name_override" not in option_keys
463 and "grpc.default_authority" not in option_keys
464 ):
465 kwargs["options"] = existing_options + (
466 ("grpc.default_authority", authority),
467 )
469 if attempt_direct_path or use_dp_interconnect:
470 target = _modify_target_for_direct_path(
471 target,
472 attempt_direct_path_xds_over_interconnect=use_dp_interconnect,
473 )
474 elif not target.startswith("google-c2p:///"):
475 clean_host = _extract_target_host(target)
476 if clean_host.endswith("-direct.googleapis.com") and len(clean_host) > len(
477 "-direct.googleapis.com"
478 ):
479 new_host = (
480 clean_host.removesuffix("-direct.googleapis.com") + ".googleapis.com"
481 )
482 target = target.replace(clean_host, new_host, 1)
484 return target, composite_credentials, kwargs
487def create_channel(
488 target,
489 credentials=None,
490 scopes=None,
491 ssl_credentials=None,
492 credentials_file=None,
493 quota_project_id=None,
494 default_scopes=None,
495 default_host=None,
496 compression=None,
497 attempt_direct_path: Optional[bool] = False,
498 attempt_direct_path_xds_over_interconnect: Optional[bool] = None,
499 **kwargs,
500):
501 """Create a secure channel with credentials.
503 Args:
504 target (str): The target service address in the format 'hostname:port'.
505 credentials (google.auth.credentials.Credentials): The credentials. If
506 not specified, then this function will attempt to ascertain the
507 credentials from the environment using :func:`google.auth.default`.
508 scopes (Sequence[str]): A optional list of scopes needed for this
509 service. These are only used when credentials are not specified and
510 are passed to :func:`google.auth.default`.
511 ssl_credentials (grpc.ChannelCredentials): Optional SSL channel
512 credentials. This can be used to specify different certificates.
513 credentials_file (str): A file with credentials that can be loaded with
514 :func:`google.auth.load_credentials_from_file`. This argument is
515 mutually exclusive with credentials.
517 .. warning::
518 Important: If you accept a credential configuration (credential JSON/File/Stream)
519 from an external source for authentication to Google Cloud Platform, you must
520 validate it before providing it to any Google API or client library. Providing an
521 unvalidated credential configuration to Google APIs or libraries can compromise
522 the security of your systems and data. For more information, refer to
523 `Validate credential configurations from external sources`_.
525 .. _Validate credential configurations from external sources:
527 https://cloud.google.com/docs/authentication/external/externally-sourced-credentials
528 quota_project_id (str): An optional project to use for billing and quota.
529 default_scopes (Sequence[str]): Default scopes passed by a Google client
530 library. Use 'scopes' for user-defined scopes.
531 default_host (str): The default endpoint. e.g., "pubsub.googleapis.com".
532 compression (grpc.Compression): An optional value indicating the
533 compression method to be used over the lifetime of the channel.
534 attempt_direct_path (Optional[bool]): If set, Direct Path will be attempted
535 when the request is made. Direct Path is only available within a Google
536 Compute Engine (GCE) environment and provides a proxyless connection
537 which increases the available throughput, reduces latency, and increases
538 reliability. Note:
540 - This argument should only be set in a GCE environment and for Services
541 that are known to support Direct Path.
542 - If this argument is set outside of GCE, then this request will fail
543 unless the back-end service happens to have configured fall-back to DNS.
544 - If the request causes a `ServiceUnavailable` response, it is recommended
545 that the client repeat the request with `attempt_direct_path` set to
546 `False` as the Service may not support Direct Path.
547 - Using `ssl_credentials` with `attempt_direct_path` set to `True` will
548 result in `ValueError` as this combination is not yet supported.
549 attempt_direct_path_xds_over_interconnect (Optional[bool]): If set,
550 DirectPath over Cloud Interconnect will be attempted using standard
551 TLS credentials and ``?force-xds`` C2P target resolution. If None
552 (default), the setting is resolved from the
553 ``GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT``
554 environment variable.
556 kwargs: Additional key-word args passed to
557 :func:`grpc.secure_channel`.
559 Returns:
560 grpc.Channel: The created channel.
562 Raises:
563 google.api_core.DuplicateCredentialArgs: If both a credentials object
564 and credentials_file are passed.
565 ValueError: If `ssl_credentials` is set and `attempt_direct_path` is
566 set to `True` without `attempt_direct_path_xds_over_interconnect`.
567 """
569 target, composite_credentials, kwargs = _setup_direct_path_and_credentials(
570 target=target,
571 credentials=credentials,
572 scopes=scopes,
573 ssl_credentials=ssl_credentials,
574 credentials_file=credentials_file,
575 quota_project_id=quota_project_id,
576 default_scopes=default_scopes,
577 default_host=default_host,
578 attempt_direct_path=attempt_direct_path,
579 attempt_direct_path_xds_over_interconnect=attempt_direct_path_xds_over_interconnect,
580 **kwargs,
581 )
583 return grpc.secure_channel(
584 target, composite_credentials, compression=compression, **kwargs
585 )
588def _modify_target_for_direct_path(
589 target: str,
590 attempt_direct_path_xds_over_interconnect: Optional[bool] = False,
591) -> str:
592 """
593 Given a target, return a modified version which is compatible with Direct Path.
595 Args:
596 target (str): The target service address in the format 'hostname[:port]' or
597 'dns://hostname[:port]'.
598 attempt_direct_path_xds_over_interconnect (Optional[bool]): Whether to
599 append ``?force-xds`` for DirectPath over Cloud Interconnect.
601 Returns:
602 target (str): The target service address which is converted into a format compatible with Direct Path.
603 If the target contains `dns:///` or does not contain `:///`, the target will be converted in
604 a format compatible with Direct Path; otherwise the original target will be returned as the
605 original target may already denote Direct Path.
606 """
608 # Strip standard URI scheme prefixes ("dns:///", "https://", "http://") if
609 # `attempt_direct_path` is enabled, as the Direct Path prefix `google-c2p:///`
610 # will be used instead.
611 for scheme_prefix in ("dns:///", "https://", "http://"):
612 if target.startswith(scheme_prefix):
613 target = target.removeprefix(scheme_prefix)
614 break
616 direct_path_separator = ":///"
617 if direct_path_separator not in target:
618 if "?" in target:
619 host_part, query_part = target.split("?", 1)
620 target_without_port = host_part.split("/")[0].split(":")[0]
621 target = (
622 f"google-c2p{direct_path_separator}{target_without_port}?{query_part}"
623 )
624 else:
625 target_without_port = target.split("/")[0].split(":")[0]
626 # Modify the target to use Direct Path by adding the `google-c2p:///` prefix
627 target = f"google-c2p{direct_path_separator}{target_without_port}"
629 if attempt_direct_path_xds_over_interconnect and target.startswith(
630 "google-c2p:///"
631 ):
632 if "force-xds" not in target:
633 separator = "&" if "?" in target else "?"
634 target = f"{target}{separator}force-xds"
635 return target
638def apply_channel_interceptors(
639 channel: grpc.Channel,
640 interceptors: (
641 Sequence[ClientInterceptor | Callable[[grpc.Channel], grpc.Channel]] | None
642 ) = None,
643) -> grpc.Channel:
644 """Applies client interceptors or channel-intercepting callables to a gRPC channel.
646 Executes in reverse order so the first interceptor in the sequence becomes the
647 outermost layer on outbound requests and the innermost layer on inbound responses,
648 aligning with the behavior of ``grpc.intercept_channel``.
650 Args:
651 channel (grpc.Channel): The channel to intercept.
652 interceptors (Optional[Sequence[Union[ClientInterceptor, Callable[[grpc.Channel], grpc.Channel]]]]):
653 Additional interceptors (or callables that apply interceptors) to apply to the gRPC channel.
655 Returns:
656 grpc.Channel: The intercepted channel, or the original channel if no
657 interceptors were provided.
659 Raises:
660 TypeError: If an item in ``interceptors`` is neither a gRPC ClientInterceptor
661 nor a Callable[[Channel], Channel].
662 """
663 if not interceptors:
664 return channel
666 modified_channel = channel
667 # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors)
668 for interceptor in reversed(list(interceptors)):
669 if isinstance(interceptor, _CLIENT_INTERCEPTOR_CLASSES):
670 modified_channel = grpc.intercept_channel(modified_channel, interceptor)
671 elif callable(interceptor):
672 interceptor_callable = cast(
673 Callable[[grpc.Channel], grpc.Channel], interceptor
674 )
675 modified_channel = interceptor_callable(modified_channel)
676 else:
677 raise TypeError(
678 f"Expected ClientInterceptor or Callable[[Channel], Channel], got {type(interceptor).__name__}"
679 )
681 return modified_channel
684_MethodCall = collections.namedtuple(
685 "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
686)
688_ChannelRequest = collections.namedtuple("_ChannelRequest", ("method", "request"))
691class _CallableStub(object):
692 """Stub for the grpc.*MultiCallable interfaces."""
694 def __init__(self, method, channel):
695 self._method = method
696 self._channel = channel
697 self.response = None
698 """Union[protobuf.Message, Callable[protobuf.Message], exception]:
699 The response to give when invoking this callable. If this is a
700 callable, it will be invoked with the request protobuf. If it's an
701 exception, the exception will be raised when this is invoked.
702 """
703 self.responses = None
704 """Iterator[
705 Union[protobuf.Message, Callable[protobuf.Message], exception]]:
706 An iterator of responses. If specified, self.response will be populated
707 on each invocation by calling ``next(self.responses)``."""
708 self.requests = []
709 """List[protobuf.Message]: All requests sent to this callable."""
710 self.calls = []
711 """List[Tuple]: All invocations of this callable. Each tuple is the
712 request, timeout, metadata, compression, and credentials."""
714 def __call__(
715 self, request, timeout=None, metadata=None, credentials=None, compression=None
716 ):
717 self._channel.requests.append(_ChannelRequest(self._method, request))
718 self.calls.append(
719 _MethodCall(request, timeout, metadata, credentials, compression)
720 )
721 self.requests.append(request)
723 response = self.response
724 if self.responses is not None:
725 if response is None:
726 response = next(self.responses)
727 else:
728 raise ValueError(
729 "{method}.response and {method}.responses are mutually "
730 "exclusive.".format(method=self._method)
731 )
733 if callable(response):
734 return response(request)
736 if isinstance(response, Exception):
737 raise response
739 if response is not None:
740 return response
742 raise ValueError('Method stub for "{}" has no response.'.format(self._method))
745def _simplify_method_name(method):
746 """Simplifies a gRPC method name.
748 When gRPC invokes the channel to create a callable, it gives a full
749 method name like "/google.pubsub.v1.Publisher/CreateTopic". This
750 returns just the name of the method, in this case "CreateTopic".
752 Args:
753 method (str): The name of the method.
755 Returns:
756 str: The simplified name of the method.
757 """
758 return method.rsplit("/", 1).pop()
761class ChannelStub(grpc.Channel):
762 """A testing stub for the grpc.Channel interface.
764 This can be used to test any client that eventually uses a gRPC channel
765 to communicate. By passing in a channel stub, you can configure which
766 responses are returned and track which requests are made.
768 For example:
770 .. code-block:: python
772 channel_stub = grpc_helpers.ChannelStub()
773 client = FooClient(channel=channel_stub)
775 channel_stub.GetFoo.response = foo_pb2.Foo(name='bar')
777 foo = client.get_foo(labels=['baz'])
779 assert foo.name == 'bar'
780 assert channel_stub.GetFoo.requests[0].labels = ['baz']
782 Each method on the stub can be accessed and configured on the channel.
783 Here's some examples of various configurations:
785 .. code-block:: python
787 # Return a basic response:
789 channel_stub.GetFoo.response = foo_pb2.Foo(name='bar')
790 assert client.get_foo().name == 'bar'
792 # Raise an exception:
793 channel_stub.GetFoo.response = NotFound('...')
795 with pytest.raises(NotFound):
796 client.get_foo()
798 # Use a sequence of responses:
799 channel_stub.GetFoo.responses = iter([
800 foo_pb2.Foo(name='bar'),
801 foo_pb2.Foo(name='baz'),
802 ])
804 assert client.get_foo().name == 'bar'
805 assert client.get_foo().name == 'baz'
807 # Use a callable
809 def on_get_foo(request):
810 return foo_pb2.Foo(name='bar' + request.id)
812 channel_stub.GetFoo.response = on_get_foo
814 assert client.get_foo(id='123').name == 'bar123'
815 """
817 def __init__(self, responses=[]):
818 self.requests = []
819 """Sequence[Tuple[str, protobuf.Message]]: A list of all requests made
820 on this channel in order. The tuple is of method name, request
821 message."""
822 self._method_stubs = {}
824 def _stub_for_method(self, method):
825 method = _simplify_method_name(method)
826 self._method_stubs[method] = _CallableStub(method, self)
827 return self._method_stubs[method]
829 def __getattr__(self, key):
830 try:
831 return self._method_stubs[key]
832 except KeyError:
833 raise AttributeError
835 def unary_unary(
836 self,
837 method,
838 request_serializer=None,
839 response_deserializer=None,
840 _registered_method=False,
841 ):
842 """grpc.Channel.unary_unary implementation."""
843 return self._stub_for_method(method)
845 def unary_stream(
846 self,
847 method,
848 request_serializer=None,
849 response_deserializer=None,
850 _registered_method=False,
851 ):
852 """grpc.Channel.unary_stream implementation."""
853 return self._stub_for_method(method)
855 def stream_unary(
856 self,
857 method,
858 request_serializer=None,
859 response_deserializer=None,
860 _registered_method=False,
861 ):
862 """grpc.Channel.stream_unary implementation."""
863 return self._stub_for_method(method)
865 def stream_stream(
866 self,
867 method,
868 request_serializer=None,
869 response_deserializer=None,
870 _registered_method=False,
871 ):
872 """grpc.Channel.stream_stream implementation."""
873 return self._stub_for_method(method)
875 def subscribe(self, callback, try_to_connect=False):
876 """grpc.Channel.subscribe implementation."""
877 pass
879 def unsubscribe(self, callback):
880 """grpc.Channel.unsubscribe implementation."""
881 pass
883 def close(self):
884 """grpc.Channel.close implementation."""
885 pass