src/python/grpcio/grpc/aio/_base_channel.py

   1 # Copyright 2020 The gRPC Authors
   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.
  14 """Abstract base classes for Channel objects and Multicallable objects."""
  15
  16 import abc
  17 from typing import Any, Optional
  18
  19 import grpc
  20
  21 from . import _base_call
  22 from ._typing import (DeserializingFunction, RequestIterableType,
  23                       SerializingFunction)
  24 from ._metadata import Metadata
  25
  26
  27 class UnaryUnaryMultiCallable(abc.ABC):
  28     """Enables asynchronous invocation of a unary-call RPC."""
  29
  30     @abc.abstractmethod
  31     def __call__(
  32         self,
  33         request: Any,
  34         *,
  35         timeout: Optional[float] = None,
  36         metadata: Optional[Metadata] = None,
  37         credentials: Optional[grpc.CallCredentials] = None,
  38         wait_for_ready: Optional[bool] = None,
  39         compression: Optional[grpc.Compression] = None
  40     ) -> _base_call.UnaryUnaryCall:
  41         """Asynchronously invokes the underlying RPC.
  42
  43         Args:
  44           request: The request value for the RPC.
  45           timeout: An optional duration of time in seconds to allow
  46             for the RPC.
  47           metadata: Optional :term:`metadata` to be transmitted to the
  48             service-side of the RPC.
  49           credentials: An optional CallCredentials for the RPC. Only valid for
  50             secure Channel.
  51           wait_for_ready: This is an EXPERIMENTAL argument. An optional
  52             flag to enable :term:`wait_for_ready` mechanism.
  53           compression: An element of grpc.compression, e.g.
  54             grpc.compression.Gzip. This is an EXPERIMENTAL option.
  55
  56         Returns:
  57           A UnaryUnaryCall object.
  58
  59         Raises:
  60           RpcError: Indicates that the RPC terminated with non-OK status. The
  61             raised RpcError will also be a Call for the RPC affording the RPC's
  62             metadata, status code, and details.
  63         """
  64
  65
  66 class UnaryStreamMultiCallable(abc.ABC):
  67     """Enables asynchronous invocation of a server-streaming RPC."""
  68
  69     @abc.abstractmethod
  70     def __call__(
  71         self,
  72         request: Any,
  73         *,
  74         timeout: Optional[float] = None,
  75         metadata: Optional[Metadata] = None,
  76         credentials: Optional[grpc.CallCredentials] = None,
  77         wait_for_ready: Optional[bool] = None,
  78         compression: Optional[grpc.Compression] = None
  79     ) -> _base_call.UnaryStreamCall:
  80         """Asynchronously invokes the underlying RPC.
  81
  82         Args:
  83           request: The request value for the RPC.
  84           timeout: An optional duration of time in seconds to allow
  85             for the RPC.
  86           metadata: Optional :term:`metadata` to be transmitted to the
  87             service-side of the RPC.
  88           credentials: An optional CallCredentials for the RPC. Only valid for
  89             secure Channel.
  90           wait_for_ready: This is an EXPERIMENTAL argument. An optional
  91             flag to enable :term:`wait_for_ready` mechanism.
  92           compression: An element of grpc.compression, e.g.
  93             grpc.compression.Gzip. This is an EXPERIMENTAL option.
  94
  95         Returns:
  96           A UnaryStreamCall object.
  97
  98         Raises:
  99           RpcError: Indicates that the RPC terminated with non-OK status. The
 100             raised RpcError will also be a Call for the RPC affording the RPC's
 101             metadata, status code, and details.
 102         """
 103
 104
 105 class StreamUnaryMultiCallable(abc.ABC):
 106     """Enables asynchronous invocation of a client-streaming RPC."""
 107
 108     @abc.abstractmethod
 109     def __call__(
 110         self,
 111         request_iterator: Optional[RequestIterableType] = None,
 112         timeout: Optional[float] = None,
 113         metadata: Optional[Metadata] = None,
 114         credentials: Optional[grpc.CallCredentials] = None,
 115         wait_for_ready: Optional[bool] = None,
 116         compression: Optional[grpc.Compression] = None
 117     ) -> _base_call.StreamUnaryCall:
 118         """Asynchronously invokes the underlying RPC.
 119
 120         Args:
 121           request_iterator: An optional async iterable or iterable of request
 122             messages for the RPC.
 123           timeout: An optional duration of time in seconds to allow
 124             for the RPC.
 125           metadata: Optional :term:`metadata` to be transmitted to the
 126             service-side of the RPC.
 127           credentials: An optional CallCredentials for the RPC. Only valid for
 128             secure Channel.
 129           wait_for_ready: This is an EXPERIMENTAL argument. An optional
 130             flag to enable :term:`wait_for_ready` mechanism.
 131           compression: An element of grpc.compression, e.g.
 132             grpc.compression.Gzip. This is an EXPERIMENTAL option.
 133
 134         Returns:
 135           A StreamUnaryCall object.
 136
 137         Raises:
 138           RpcError: Indicates that the RPC terminated with non-OK status. The
 139             raised RpcError will also be a Call for the RPC affording the RPC's
 140             metadata, status code, and details.
 141         """
 142
 143
 144 class StreamStreamMultiCallable(abc.ABC):
 145     """Enables asynchronous invocation of a bidirectional-streaming RPC."""
 146
 147     @abc.abstractmethod
 148     def __call__(
 149         self,
 150         request_iterator: Optional[RequestIterableType] = None,
 151         timeout: Optional[float] = None,
 152         metadata: Optional[Metadata] = None,
 153         credentials: Optional[grpc.CallCredentials] = None,
 154         wait_for_ready: Optional[bool] = None,
 155         compression: Optional[grpc.Compression] = None
 156     ) -> _base_call.StreamStreamCall:
 157         """Asynchronously invokes the underlying RPC.
 158
 159         Args:
 160           request_iterator: An optional async iterable or iterable of request
 161             messages for the RPC.
 162           timeout: An optional duration of time in seconds to allow
 163             for the RPC.
 164           metadata: Optional :term:`metadata` to be transmitted to the
 165             service-side of the RPC.
 166           credentials: An optional CallCredentials for the RPC. Only valid for
 167             secure Channel.
 168           wait_for_ready: This is an EXPERIMENTAL argument. An optional
 169             flag to enable :term:`wait_for_ready` mechanism.
 170           compression: An element of grpc.compression, e.g.
 171             grpc.compression.Gzip. This is an EXPERIMENTAL option.
 172
 173         Returns:
 174           A StreamStreamCall object.
 175
 176         Raises:
 177           RpcError: Indicates that the RPC terminated with non-OK status. The
 178             raised RpcError will also be a Call for the RPC affording the RPC's
 179             metadata, status code, and details.
 180         """
 181
 182
 183 class Channel(abc.ABC):
 184     """Enables asynchronous RPC invocation as a client.
 185
 186     Channel objects implement the Asynchronous Context Manager (aka. async
 187     with) type, although they are not supportted to be entered and exited
 188     multiple times.
 189     """
 190
 191     @abc.abstractmethod
 192     async def __aenter__(self):
 193         """Starts an asynchronous context manager.
 194
 195         Returns:
 196           Channel the channel that was instantiated.
 197         """
 198
 199     @abc.abstractmethod
 200     async def __aexit__(self, exc_type, exc_val, exc_tb):
 201         """Finishes the asynchronous context manager by closing the channel.
 202
 203         Still active RPCs will be cancelled.
 204         """
 205
 206     @abc.abstractmethod
 207     async def close(self, grace: Optional[float] = None):
 208         """Closes this Channel and releases all resources held by it.
 209
 210         This method immediately stops the channel from executing new RPCs in
 211         all cases.
 212
 213         If a grace period is specified, this method wait until all active
 214         RPCs are finshed, once the grace period is reached the ones that haven't
 215         been terminated are cancelled. If a grace period is not specified
 216         (by passing None for grace), all existing RPCs are cancelled immediately.
 217
 218         This method is idempotent.
 219         """
 220
 221     @abc.abstractmethod
 222     def get_state(self,
 223                   try_to_connect: bool = False) -> grpc.ChannelConnectivity:
 224         """Checks the connectivity state of a channel.
 225
 226         This is an EXPERIMENTAL API.
 227
 228         If the channel reaches a stable connectivity state, it is guaranteed
 229         that the return value of this function will eventually converge to that
 230         state.
 231
 232         Args:
 233           try_to_connect: a bool indicate whether the Channel should try to
 234             connect to peer or not.
 235
 236         Returns: A ChannelConnectivity object.
 237         """
 238
 239     @abc.abstractmethod
 240     async def wait_for_state_change(
 241         self,
 242         last_observed_state: grpc.ChannelConnectivity,
 243     ) -> None:
 244         """Waits for a change in connectivity state.
 245
 246         This is an EXPERIMENTAL API.
 247
 248         The function blocks until there is a change in the channel connectivity
 249         state from the "last_observed_state". If the state is already
 250         different, this function will return immediately.
 251
 252         There is an inherent race between the invocation of
 253         "Channel.wait_for_state_change" and "Channel.get_state". The state can
 254         change arbitrary many times during the race, so there is no way to
 255         observe every state transition.
 256
 257         If there is a need to put a timeout for this function, please refer to
 258         "asyncio.wait_for".
 259
 260         Args:
 261           last_observed_state: A grpc.ChannelConnectivity object representing
 262             the last known state.
 263         """
 264
 265     @abc.abstractmethod
 266     async def channel_ready(self) -> None:
 267         """Creates a coroutine that blocks until the Channel is READY."""
 268
 269     @abc.abstractmethod
 270     def unary_unary(
 271         self,
 272         method: str,
 273         request_serializer: Optional[SerializingFunction] = None,
 274         response_deserializer: Optional[DeserializingFunction] = None
 275     ) -> UnaryUnaryMultiCallable:
 276         """Creates a UnaryUnaryMultiCallable for a unary-unary method.
 277
 278         Args:
 279           method: The name of the RPC method.
 280           request_serializer: Optional :term:`serializer` for serializing the request
 281             message. Request goes unserialized in case None is passed.
 282           response_deserializer: Optional :term:`deserializer` for deserializing the
 283             response message. Response goes undeserialized in case None
 284             is passed.
 285
 286         Returns:
 287           A UnaryUnaryMultiCallable value for the named unary-unary method.
 288         """
 289
 290     @abc.abstractmethod
 291     def unary_stream(
 292         self,
 293         method: str,
 294         request_serializer: Optional[SerializingFunction] = None,
 295         response_deserializer: Optional[DeserializingFunction] = None
 296     ) -> UnaryStreamMultiCallable:
 297         """Creates a UnaryStreamMultiCallable for a unary-stream method.
 298
 299         Args:
 300           method: The name of the RPC method.
 301           request_serializer: Optional :term:`serializer` for serializing the request
 302             message. Request goes unserialized in case None is passed.
 303           response_deserializer: Optional :term:`deserializer` for deserializing the
 304             response message. Response goes undeserialized in case None
 305             is passed.
 306
 307         Returns:
 308           A UnarySteramMultiCallable value for the named unary-stream method.
 309         """
 310
 311     @abc.abstractmethod
 312     def stream_unary(
 313         self,
 314         method: str,
 315         request_serializer: Optional[SerializingFunction] = None,
 316         response_deserializer: Optional[DeserializingFunction] = None
 317     ) -> StreamUnaryMultiCallable:
 318         """Creates a StreamUnaryMultiCallable for a stream-unary method.
 319
 320         Args:
 321           method: The name of the RPC method.
 322           request_serializer: Optional :term:`serializer` for serializing the request
 323             message. Request goes unserialized in case None is passed.
 324           response_deserializer: Optional :term:`deserializer` for deserializing the
 325             response message. Response goes undeserialized in case None
 326             is passed.
 327
 328         Returns:
 329           A StreamUnaryMultiCallable value for the named stream-unary method.
 330         """
 331
 332     @abc.abstractmethod
 333     def stream_stream(
 334         self,
 335         method: str,
 336         request_serializer: Optional[SerializingFunction] = None,
 337         response_deserializer: Optional[DeserializingFunction] = None
 338     ) -> StreamStreamMultiCallable:
 339         """Creates a StreamStreamMultiCallable for a stream-stream method.
 340
 341         Args:
 342           method: The name of the RPC method.
 343           request_serializer: Optional :term:`serializer` for serializing the request
 344             message. Request goes unserialized in case None is passed.
 345           response_deserializer: Optional :term:`deserializer` for deserializing the
 346             response message. Response goes undeserialized in case None
 347             is passed.
 348
 349         Returns:
 350           A StreamStreamMultiCallable value for the named stream-stream method.
 351         """