Source code for grpc.aio._base_channel

# Copyright 2020 The gRPC Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Abstract base classes for Channel objects and Multicallable objects."""

import abc
from typing import Any, Optional

import grpc

from . import _base_call
from ._typing import DeserializingFunction
from ._typing import MetadataType
from ._typing import RequestIterableType
from ._typing import SerializingFunction


[docs]class UnaryUnaryMultiCallable(abc.ABC): """Enables asynchronous invocation of a unary-call RPC."""
[docs] @abc.abstractmethod def __call__( self, request: Any, *, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None ) -> _base_call.UnaryUnaryCall: """Asynchronously invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: This is an EXPERIMENTAL argument. An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. This is an EXPERIMENTAL option. Returns: A UnaryUnaryCall object. Raises: RpcError: Indicates that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """
[docs]class UnaryStreamMultiCallable(abc.ABC): """Enables asynchronous invocation of a server-streaming RPC."""
[docs] @abc.abstractmethod def __call__( self, request: Any, *, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None ) -> _base_call.UnaryStreamCall: """Asynchronously invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: This is an EXPERIMENTAL argument. An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. This is an EXPERIMENTAL option. Returns: A UnaryStreamCall object. Raises: RpcError: Indicates that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """
[docs]class StreamUnaryMultiCallable(abc.ABC): """Enables asynchronous invocation of a client-streaming RPC."""
[docs] @abc.abstractmethod def __call__( self, request_iterator: Optional[RequestIterableType] = None, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None ) -> _base_call.StreamUnaryCall: """Asynchronously invokes the underlying RPC. Args: request_iterator: An optional async iterable or iterable of request messages for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: This is an EXPERIMENTAL argument. An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. This is an EXPERIMENTAL option. Returns: A StreamUnaryCall object. Raises: RpcError: Indicates that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """
[docs]class StreamStreamMultiCallable(abc.ABC): """Enables asynchronous invocation of a bidirectional-streaming RPC."""
[docs] @abc.abstractmethod def __call__( self, request_iterator: Optional[RequestIterableType] = None, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None ) -> _base_call.StreamStreamCall: """Asynchronously invokes the underlying RPC. Args: request_iterator: An optional async iterable or iterable of request messages for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: This is an EXPERIMENTAL argument. An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. This is an EXPERIMENTAL option. Returns: A StreamStreamCall object. Raises: RpcError: Indicates that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """
[docs]class Channel(abc.ABC): """Enables asynchronous RPC invocation as a client. Channel objects implement the Asynchronous Context Manager (aka. async with) type, although they are not supportted to be entered and exited multiple times. """
[docs] @abc.abstractmethod async def __aenter__(self): """Starts an asynchronous context manager. Returns: Channel the channel that was instantiated. """
[docs] @abc.abstractmethod async def __aexit__(self, exc_type, exc_val, exc_tb): """Finishes the asynchronous context manager by closing the channel. Still active RPCs will be cancelled. """
[docs] @abc.abstractmethod async def close(self, grace: Optional[float] = None): """Closes this Channel and releases all resources held by it. This method immediately stops the channel from executing new RPCs in all cases. If a grace period is specified, this method wait until all active RPCs are finshed, once the grace period is reached the ones that haven't been terminated are cancelled. If a grace period is not specified (by passing None for grace), all existing RPCs are cancelled immediately. This method is idempotent. """
[docs] @abc.abstractmethod def get_state(self, try_to_connect: bool = False) -> grpc.ChannelConnectivity: """Checks the connectivity state of a channel. This is an EXPERIMENTAL API. If the channel reaches a stable connectivity state, it is guaranteed that the return value of this function will eventually converge to that state. Args: try_to_connect: a bool indicate whether the Channel should try to connect to peer or not. Returns: A ChannelConnectivity object. """
[docs] @abc.abstractmethod async def wait_for_state_change( self, last_observed_state: grpc.ChannelConnectivity, ) -> None: """Waits for a change in connectivity state. This is an EXPERIMENTAL API. The function blocks until there is a change in the channel connectivity state from the "last_observed_state". If the state is already different, this function will return immediately. There is an inherent race between the invocation of "Channel.wait_for_state_change" and "Channel.get_state". The state can change arbitrary many times during the race, so there is no way to observe every state transition. If there is a need to put a timeout for this function, please refer to "asyncio.wait_for". Args: last_observed_state: A grpc.ChannelConnectivity object representing the last known state. """
[docs] @abc.abstractmethod async def channel_ready(self) -> None: """Creates a coroutine that blocks until the Channel is READY."""
[docs] @abc.abstractmethod def unary_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None ) -> UnaryUnaryMultiCallable: """Creates a UnaryUnaryMultiCallable for a unary-unary method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. Returns: A UnaryUnaryMultiCallable value for the named unary-unary method. """
[docs] @abc.abstractmethod def unary_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None ) -> UnaryStreamMultiCallable: """Creates a UnaryStreamMultiCallable for a unary-stream method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. Returns: A UnarySteramMultiCallable value for the named unary-stream method. """
[docs] @abc.abstractmethod def stream_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None ) -> StreamUnaryMultiCallable: """Creates a StreamUnaryMultiCallable for a stream-unary method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. Returns: A StreamUnaryMultiCallable value for the named stream-unary method. """
[docs] @abc.abstractmethod def stream_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None ) -> StreamStreamMultiCallable: """Creates a StreamStreamMultiCallable for a stream-stream method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. Returns: A StreamStreamMultiCallable value for the named stream-stream method. """