Fill-in missing interfaces & improve rst template

doc/python/sphinx/grpc_asyncio.rst

@@ -6,12 +6,12 @@ gRPC AsyncIO API
 Overview
 --------
 
-gRPC AsyncIO API is a new version of gRPC Python whose architecture is
+gRPC AsyncIO API is the **new version** of gRPC Python whose architecture is
 tailored to AsyncIO. Underlying, it is using C-Core's callback API, and
 replaced all IO operations with methods provided by the AsyncIO library.
 
 This stack currently is under active development. Feel free to offer
-suggestions by opening issues on `grpc/grpc <https://github.com/grpc/grpc>`_.
+suggestions by opening issues on our GitHub repo `grpc/grpc <https://github.com/grpc/grpc>`_.
 
 The design doc can be found here as `gRFC <https://github.com/grpc/proposal/pull/155>`_.
 
@@ -19,6 +19,11 @@ The design doc can be found here as `gRFC <https://github.com/grpc/proposal/pull
 Module Contents
 ---------------
 
+Turn-On AsyncIO Mode
+^^^^^^^^^^^^^^^^^^^^
+
+.. autofunction:: init_grpc_aio
+
 
 Create Client
 ^^^^^^^^^^^^^
@@ -27,18 +32,18 @@ Create Client
 .. autofunction:: secure_channel
 
 
-Create Server
-^^^^^^^^^^^^^
-
-.. autofunction:: server
-
-
 Channel Object
 ^^^^^^^^^^^^^^
 
 .. autoclass:: Channel
 
 
+Create Server
+^^^^^^^^^^^^^
+
+.. autofunction:: server
+
+
 Server Object
 ^^^^^^^^^^^^^
 
@@ -55,6 +60,12 @@ gRPC Exceptions
 .. autoexception:: AioRpcError
 
 
+Shared Context
+^^^^^^^^^^^^^^^^^^^^
+
+.. autoclass:: RpcContext
+
+
 Client-Side Context
 ^^^^^^^^^^^^^^^^^^^^^^^
 
@@ -65,6 +76,12 @@ Client-Side Context
 .. autoclass:: StreamStreamCall
 
 
+Server-Side Context
+^^^^^^^^^^^^^^^^^^^^^^^
+
+.. autoclass:: ServicerContext
+
+
 Client-Side Interceptor
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
@@ -78,31 +95,10 @@ Client-Side Interceptor
 .. .. autoclass:: ServicerContext
 
 
-.. Service-Side Interceptor
-.. ^^^^^^^^^^^^^^^^^^^^^^^^
-
-.. .. autoclass:: ServerInterceptor
-
-
 Multi-Callable Interfaces
 ^^^^^^^^^^^^^^^^^^^^^^^^^
 
-.. autoclass:: UnaryUnaryMultiCallable()
+.. autoclass:: UnaryUnaryMultiCallable
-    :undoc-members: __init__
 .. autoclass:: UnaryStreamMultiCallable()
 .. autoclass:: StreamUnaryMultiCallable()
 .. autoclass:: StreamStreamMultiCallable()
-
-
-.. Future Interfaces
-.. ^^^^^^^^^^^^^^^^^
-
-.. .. autoexception:: FutureTimeoutError
-.. .. autoexception:: FutureCancelledError
-.. .. autoclass:: Future
-
-
-.. Compression
-.. ^^^^^^^^^^^
-
-.. .. autoclass:: Compression

src/python/grpcio/grpc/_cython/_cygrpc/aio/grpc_aio.pyx.pxi

@@ -17,6 +17,14 @@ cdef bint _grpc_aio_initialized = 0
 
 
 def init_grpc_aio():
+    """Turn-on AsyncIO mode for gRPC Python.
+
+    This function enables AsyncIO IO manager and disables threading for entire
+    process. After this point, there should not be blocking calls unless it is
+    taken cared by AsyncIO.
+
+    This function is idempotent.
+    """
     global _grpc_aio_initialized
 
     if _grpc_aio_initialized:

src/python/grpcio/grpc/experimental/aio/__init__.py

@@ -20,74 +20,49 @@ created. AsyncIO doesn't provide thread safety for most of its APIs.
 from typing import Any, Optional, Sequence, Tuple
 
 import grpc
-from grpc._cython.cygrpc import (EOF, AbortError, BaseError, UsageError,
+from grpc._cython.cygrpc import (EOF, AbortError, BaseError, InternalError,
-                                 InternalError, init_grpc_aio)
+                                 UsageError, init_grpc_aio)
 
-from ._base_call import Call, RpcContext, UnaryStreamCall, UnaryUnaryCall, StreamUnaryCall, StreamStreamCall
+from ._base_call import (Call, RpcContext, StreamStreamCall, StreamUnaryCall,
+                         UnaryStreamCall, UnaryUnaryCall)
+from ._base_channel import (Channel, StreamStreamMultiCallable,
+                            StreamUnaryMultiCallable, UnaryStreamMultiCallable,
+                            UnaryUnaryMultiCallable)
 from ._call import AioRpcError
-from ._channel import Channel, UnaryUnaryMultiCallable, UnaryStreamMultiCallable, StreamUnaryMultiCallable, StreamStreamMultiCallable
 from ._interceptor import (ClientCallDetails, InterceptedUnaryUnaryCall,
                            UnaryUnaryClientInterceptor)
-from ._server import Server, server
+from ._server import server
+from ._base_server import Server, ServicerContext
 from ._typing import ChannelArgumentType
-
+from ._channel import insecure_channel, secure_channel
-
-def insecure_channel(
-        target: str,
-        options: Optional[ChannelArgumentType] = None,
-        compression: Optional[grpc.Compression] = None,
-        interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]] = None):
-    """Creates an insecure asynchronous Channel to a server.
-
-    Args:
-      target: The server address
-      options: An optional list of key-value pairs (channel args
-        in gRPC Core runtime) to configure the channel.
-      compression: An optional value indicating the compression method to be
-        used over the lifetime of the channel. This is an EXPERIMENTAL option.
-      interceptors: An optional sequence of interceptors that will be executed for
-        any call executed with this channel.
-
-    Returns:
-      A Channel.
-    """
-    return Channel(target, () if options is None else options, None,
-                   compression, interceptors)
-
-
-def secure_channel(
-        target: str,
-        credentials: grpc.ChannelCredentials,
-        options: Optional[ChannelArgumentType] = None,
-        compression: Optional[grpc.Compression] = None,
-        interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]] = None):
-    """Creates a secure asynchronous Channel to a server.
-
-    Args:
-      target: The server address.
-      credentials: A ChannelCredentials instance.
-      options: An optional list of key-value pairs (channel args
-        in gRPC Core runtime) to configure the channel.
-      compression: An optional value indicating the compression method to be
-        used over the lifetime of the channel. This is an EXPERIMENTAL option.
-      interceptors: An optional sequence of interceptors that will be executed for
-        any call executed with this channel.
-
-    Returns:
-      An aio.Channel.
-    """
-    return Channel(target, () if options is None else options,
-                   credentials._credentials, compression, interceptors)
-
 
 ###################################  __all__  #################################
 
-__all__ = ('AioRpcError', 'RpcContext', 'Call', 'UnaryUnaryCall',
+__all__ = (
-           'UnaryStreamCall', 'StreamUnaryCall', 'StreamStreamCall',
+    'AioRpcError',
-           'init_grpc_aio', 'Channel', 'UnaryUnaryMultiCallable',
+    'RpcContext',
-           'UnaryStreamMultiCallable', 'StreamUnaryMultiCallable',
+    'Call',
-           'StreamStreamMultiCallable', 'ClientCallDetails',
+    'UnaryUnaryCall',
-           'UnaryUnaryClientInterceptor', 'InterceptedUnaryUnaryCall',
+    'UnaryStreamCall',
-           'insecure_channel', 'server', 'Server', 'EOF', 'secure_channel',
+    'StreamUnaryCall',
-           'AbortError', 'BaseError', 'UsageError'
+    'StreamStreamCall',
-           'InternalError')
+    'init_grpc_aio',
+    'Channel',
+    'UnaryUnaryMultiCallable',
+    'UnaryStreamMultiCallable',
+    'StreamUnaryMultiCallable',
+    'StreamStreamMultiCallable',
+    'ClientCallDetails',
+    'UnaryUnaryClientInterceptor',
+    'InterceptedUnaryUnaryCall',
+    'insecure_channel',
+    'server',
+    'Server',
+    'ServicerContext',
+    'EOF',
+    'secure_channel',
+    'AbortError',
+    'BaseError',
+    'UsageError',
+    'InternalError',
+)

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

@@ -0,0 +1,339 @@
+# 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, AsyncIterable, Optional, Sequence
+
+import grpc
+
+from . import _base_call
+from ._typing import (ChannelArgumentType, DeserializingFunction, MetadataType,
+                      SerializingFunction)
+
+_IMMUTABLE_EMPTY_TUPLE = tuple()
+
+
+class UnaryUnaryMultiCallable(abc.ABC):
+    """Factory an asynchronous unary-unary RPC stub call from client-side."""
+
+    @abc.abstractmethod
+    def __call__(self,
+                 request: Any,
+                 *,
+                 timeout: Optional[float] = None,
+                 metadata: Optional[MetadataType] = _IMMUTABLE_EMPTY_TUPLE,
+                 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 wait for ready mechanism
+          compression: An element of grpc.compression, e.g.
+            grpc.compression.Gzip. This is an EXPERIMENTAL option.
+
+        Returns:
+          A Call object instance which is an awaitable object.
+
+        Raises:
+          RpcError: Indicating 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.
+        """
+
+
+class UnaryStreamMultiCallable(abc.ABC):
+    """Affords invoking a unary-stream RPC from client-side in an asynchronous way."""
+
+    @abc.abstractmethod
+    def __call__(self,
+                 request: Any,
+                 *,
+                 timeout: Optional[float] = None,
+                 metadata: Optional[MetadataType] = _IMMUTABLE_EMPTY_TUPLE,
+                 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 wait for ready mechanism
+          compression: An element of grpc.compression, e.g.
+            grpc.compression.Gzip. This is an EXPERIMENTAL option.
+
+        Returns:
+          A Call object instance which is an awaitable object.
+        """
+
+
+class StreamUnaryMultiCallable(abc.ABC):
+    """Affords invoking a stream-unary RPC from client-side in an asynchronous way."""
+
+    @abc.abstractmethod
+    def __call__(self,
+                 request_async_iterator: Optional[AsyncIterable[Any]] = None,
+                 timeout: Optional[float] = None,
+                 metadata: Optional[MetadataType] = _IMMUTABLE_EMPTY_TUPLE,
+                 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: 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 wait for ready mechanism
+          compression: An element of grpc.compression, e.g.
+            grpc.compression.Gzip. This is an EXPERIMENTAL option.
+
+        Returns:
+          A Call object instance which is an awaitable object.
+
+        Raises:
+          RpcError: Indicating 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.
+        """
+
+
+class StreamStreamMultiCallable(abc.ABC):
+    """Affords invoking a stream-stream RPC from client-side in an asynchronous way."""
+
+    @abc.abstractmethod
+    def __call__(self,
+                 request_async_iterator: Optional[AsyncIterable[Any]] = None,
+                 timeout: Optional[float] = None,
+                 metadata: Optional[MetadataType] = _IMMUTABLE_EMPTY_TUPLE,
+                 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: 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 wait for ready mechanism
+          compression: An element of grpc.compression, e.g.
+            grpc.compression.Gzip. This is an EXPERIMENTAL option.
+
+        Returns:
+          A Call object instance which is an awaitable object.
+
+        Raises:
+          RpcError: Indicating 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.
+        """
+
+
+class Channel(abc.ABC):
+    """Asynchronous Channel implementation.
+
+    A cygrpc.AioChannel-backed implementation.
+    """
+
+    @abc.abstractmethod
+    async def __aenter__(self):
+        """Starts an asynchronous context manager.
+
+        Returns:
+          Channel the channel that was instantiated.
+        """
+
+    @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.
+        """
+
+    @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.
+        """
+
+    @abc.abstractmethod
+    def get_state(self,
+                  try_to_connect: bool = False) -> grpc.ChannelConnectivity:
+        """Check 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.
+        """
+
+    @abc.abstractmethod
+    async def wait_for_state_change(
+            self,
+            last_observed_state: grpc.ChannelConnectivity,
+    ) -> None:
+        """Wait 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 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.
+        """
+
+    @abc.abstractmethod
+    async def channel_ready(self) -> None:
+        """Creates a coroutine that blocks until the Channel is READY."""
+
+    @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 behaviour for serializing the request
+            message. Request goes unserialized in case None is passed.
+          response_deserializer: Optional behaviour for deserializing the
+            response message. Response goes undeserialized in case None
+            is passed.
+
+        Returns:
+          A UnaryUnaryMultiCallable value for the named unary-unary method.
+        """
+
+    @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 behaviour for serializing the request
+            message. Request goes unserialized in case None is passed.
+          response_deserializer: Optional behaviour for deserializing the
+            response message. Response goes undeserialized in case None
+            is passed.
+
+        Returns:
+          A UnarySteramMultiCallable value for the named unary-stream method.
+        """
+
+    @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 behaviour for serializing the request
+            message. Request goes unserialized in case None is passed.
+          response_deserializer: Optional behaviour for deserializing the
+            response message. Response goes undeserialized in case None
+            is passed.
+
+        Returns:
+          A StreamUnaryMultiCallable value for the named stream-unary method.
+        """
+
+    @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 behaviour for serializing the request
+            message. Request goes unserialized in case None is passed.
+          response_deserializer: Optional behaviour for deserializing the
+            response message. Response goes undeserialized in case None
+            is passed.
+
+        Returns:
+          A StreamStreamMultiCallable value for the named stream-stream method.
+        """

src/python/grpcio/grpc/experimental/aio/_base_server.py

@@ -0,0 +1,241 @@
+# 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 server-side classes."""
+
+import abc
+from concurrent.futures import Executor
+from typing import Any, Optional, Sequence, NoReturn
+
+import grpc
+
+from ._typing import ChannelArgumentType, MetadataType, RequestType, ResponseType
+
+
+class Server(abc.ABC):
+    """Serves RPCs."""
+
+    @abc.abstractmethod
+    def add_generic_rpc_handlers(
+            self,
+            generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]) -> None:
+        """Registers GenericRpcHandlers with this Server.
+
+        This method is only safe to call before the server is started.
+
+        Args:
+          generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
+          used to service RPCs.
+        """
+
+    @abc.abstractmethod
+    def add_insecure_port(self, address: str) -> int:
+        """Opens an insecure port for accepting RPCs.
+
+        This method may only be called before starting the server.
+
+        Args:
+          address: The address for which to open a port. If the port is 0,
+            or not specified in the address, then the gRPC runtime will choose a port.
+
+        Returns:
+          An integer port on which the server will accept RPC requests.
+        """
+
+    @abc.abstractmethod
+    def add_secure_port(self, address: str,
+                        server_credentials: grpc.ServerCredentials) -> int:
+        """Opens a secure port for accepting RPCs.
+
+        This method may only be called before starting the server.
+
+        Args:
+          address: The address for which to open a port.
+            if the port is 0, or not specified in the address, then the gRPC
+            runtime will choose a port.
+          server_credentials: A ServerCredentials object.
+
+        Returns:
+          An integer port on which the server will accept RPC requests.
+        """
+
+    @abc.abstractmethod
+    async def start(self) -> None:
+        """Starts this Server.
+
+        This method may only be called once. (i.e. it is not idempotent).
+        """
+
+    @abc.abstractmethod
+    async def stop(self, grace: Optional[float]) -> None:
+        """Stops this Server.
+
+        This method immediately stops the server from servicing new RPCs in
+        all cases.
+
+        If a grace period is specified, this method returns immediately and all
+        RPCs active at the end of the grace period are aborted. If a grace
+        period is not specified (by passing None for grace), all existing RPCs
+        are aborted immediately and this method blocks until the last RPC
+        handler terminates.
+
+        This method is idempotent and may be called at any time. Passing a
+        smaller grace value in a subsequent call will have the effect of
+        stopping the Server sooner (passing None will have the effect of
+        stopping the server immediately). Passing a larger grace value in a
+        subsequent call will not have the effect of stopping the server later
+        (i.e. the most restrictive grace value is used).
+
+        Args:
+          grace: A duration of time in seconds or None.
+        """
+
+    @abc.abstractmethod
+    async def wait_for_termination(self,
+                                   timeout: Optional[float] = None) -> bool:
+        """Block current coroutine until the server stops.
+
+        This is an EXPERIMENTAL API.
+
+        The wait will not consume computational resources during blocking, and
+        it will block until one of the two following conditions are met:
+
+        1) The server is stopped or terminated;
+        2) A timeout occurs if timeout is not `None`.
+
+        The timeout argument works in the same way as `threading.Event.wait()`.
+        https://docs.python.org/3/library/threading.html#threading.Event.wait
+
+        Args:
+          timeout: A floating point number specifying a timeout for the
+            operation in seconds.
+
+        Returns:
+          A bool indicates if the operation times out.
+        """
+
+
+class ServicerContext(abc.ABC):
+    """A context object passed to method implementations."""
+
+    async def read(self) -> RequestType:
+        """Reads one message from the RPC.
+
+        Only one read operation is allowed simultaneously. Mixing new streaming API and old
+        streaming API will resulted in undefined behavior.
+
+        Returns:
+          A response message of the RPC.
+
+        Raises:
+          An RpcError exception if the read failed.
+        """
+
+    async def write(self, message: ResponseType) -> None:
+        """Writes one message to the RPC.
+
+        Only one write operation is allowed simultaneously. Mixing new streaming API and old
+        streaming API will resulted in undefined behavior.
+
+        Raises:
+          An RpcError exception if the write failed.
+        """
+
+    async def send_initial_metadata(self,
+                                    initial_metadata: MetadataType) -> None:
+        """Sends the initial metadata value to the client.
+
+        This method need not be called by implementations if they have no
+        metadata to add to what the gRPC runtime will transmit.
+
+        Args:
+          initial_metadata: The initial :term:`metadata`.
+        """
+
+    async def abort(self, code: grpc.StatusCode, details: str,
+                    trailing_metadata: MetadataType) -> NoReturn:
+        """Raises an exception to terminate the RPC with a non-OK status.
+
+        The code and details passed as arguments will supercede any existing
+        ones.
+
+        Args:
+          code: A StatusCode object to be sent to the client.
+            It must not be StatusCode.OK.
+          details: A UTF-8-encodable string to be sent to the client upon
+            termination of the RPC.
+          trailing_metadata: A sequence of tuple represents the trailing
+            :term:`metadata`.
+
+        Raises:
+          Exception: An exception is always raised to signal the abortion the
+            RPC to the gRPC runtime.
+        """
+
+    async def set_trailing_metadata(self,
+                                    trailing_metadata: MetadataType) -> None:
+        """Sends the trailing metadata for the RPC.
+
+        This method need not be called by implementations if they have no
+        metadata to add to what the gRPC runtime will transmit.
+
+        Args:
+          trailing_metadata: The trailing :term:`metadata`.
+        """
+
+    def invocation_metadata(self) -> Optional[MetadataType]:
+        """Accesses the metadata from the sent by the client.
+
+        Returns:
+          The invocation :term:`metadata`.
+        """
+
+    def set_code(self, code: grpc.StatusCode) -> None:
+        """Sets the value to be used as status code upon RPC completion.
+
+        This method need not be called by method implementations if they wish
+        the gRPC runtime to determine the status code of the RPC.
+
+        Args:
+          code: A StatusCode object to be sent to the client.
+        """
+
+    def set_details(self, details: str) -> None:
+        """Sets the value to be used as detail string upon RPC completion.
+
+        This method need not be called by method implementations if they have
+        no details to transmit.
+
+        Args:
+          details: A UTF-8-encodable string to be sent to the client upon
+            termination of the RPC.
+        """
+
+    def set_compression(self, compression: grpc.Compression) -> None:
+        """Set the compression algorithm to be used for the entire call.
+
+        This is an EXPERIMENTAL method.
+
+        Args:
+          compression: An element of grpc.compression, e.g.
+            grpc.compression.Gzip.
+        """
+
+    def disable_next_message_compression(self) -> None:
+        """Disables compression for the next response message.
+
+        This is an EXPERIMENTAL method.
+
+        This method will override any compression configuration set during
+        server creation or set on the call.
+        """

src/python/grpcio/grpc/experimental/aio/_channel.py

@@ -21,7 +21,7 @@ import grpc
 from grpc import _common, _compression, _grpcio_metadata
 from grpc._cython import cygrpc
 
-from . import _base_call
+from . import _base_call, _base_channel
 from ._call import (StreamStreamCall, StreamUnaryCall, UnaryStreamCall,
                     UnaryUnaryCall)
 from ._interceptor import (InterceptedUnaryUnaryCall,
@@ -86,8 +86,8 @@ class _BaseMultiCallable:
         self._interceptors = interceptors
 
 
-class UnaryUnaryMultiCallable(_BaseMultiCallable):
+class UnaryUnaryMultiCallable(_BaseMultiCallable,
-    """Factory an asynchronous unary-unary RPC stub call from client-side."""
+                              _base_channel.UnaryUnaryMultiCallable):
 
     def __call__(self,
                  request: Any,
@@ -98,29 +98,6 @@ class UnaryUnaryMultiCallable(_BaseMultiCallable):
                  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 wait for ready mechanism
-          compression: An element of grpc.compression, e.g.
-            grpc.compression.Gzip. This is an EXPERIMENTAL option.
-
-        Returns:
-          A Call object instance which is an awaitable object.
-
-        Raises:
-          RpcError: Indicating 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.
-        """
         if compression:
             metadata = _compression.augment_metadata(metadata, compression)
 
@@ -140,8 +117,8 @@ class UnaryUnaryMultiCallable(_BaseMultiCallable):
         return call
 
 
-class UnaryStreamMultiCallable(_BaseMultiCallable):
+class UnaryStreamMultiCallable(_BaseMultiCallable,
-    """Affords invoking a unary-stream RPC from client-side in an asynchronous way."""
+                               _base_channel.UnaryStreamMultiCallable):
 
     def __call__(self,
                  request: Any,
@@ -152,24 +129,6 @@ class UnaryStreamMultiCallable(_BaseMultiCallable):
                  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 wait for ready mechanism
-          compression: An element of grpc.compression, e.g.
-            grpc.compression.Gzip. This is an EXPERIMENTAL option.
-
-        Returns:
-          A Call object instance which is an awaitable object.
-        """
         if compression:
             metadata = _compression.augment_metadata(metadata, compression)
 
@@ -183,8 +142,8 @@ class UnaryStreamMultiCallable(_BaseMultiCallable):
         return call
 
 
-class StreamUnaryMultiCallable(_BaseMultiCallable):
+class StreamUnaryMultiCallable(_BaseMultiCallable,
-    """Affords invoking a stream-unary RPC from client-side in an asynchronous way."""
+                               _base_channel.StreamUnaryMultiCallable):
 
     def __call__(self,
                  request_async_iterator: Optional[AsyncIterable[Any]] = None,
@@ -194,29 +153,6 @@ class StreamUnaryMultiCallable(_BaseMultiCallable):
                  wait_for_ready: Optional[bool] = None,
                  compression: Optional[grpc.Compression] = None
                 ) -> _base_call.StreamUnaryCall:
-        """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 wait for ready mechanism
-          compression: An element of grpc.compression, e.g.
-            grpc.compression.Gzip. This is an EXPERIMENTAL option.
-
-        Returns:
-          A Call object instance which is an awaitable object.
-
-        Raises:
-          RpcError: Indicating 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.
-        """
         if compression:
             metadata = _compression.augment_metadata(metadata, compression)
 
@@ -230,8 +166,8 @@ class StreamUnaryMultiCallable(_BaseMultiCallable):
         return call
 
 
-class StreamStreamMultiCallable(_BaseMultiCallable):
+class StreamStreamMultiCallable(_BaseMultiCallable,
-    """Affords invoking a stream-stream RPC from client-side in an asynchronous way."""
+                                _base_channel.StreamStreamMultiCallable):
 
     def __call__(self,
                  request_async_iterator: Optional[AsyncIterable[Any]] = None,
@@ -241,29 +177,6 @@ class StreamStreamMultiCallable(_BaseMultiCallable):
                  wait_for_ready: Optional[bool] = None,
                  compression: Optional[grpc.Compression] = None
                 ) -> _base_call.StreamStreamCall:
-        """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 wait for ready mechanism
-          compression: An element of grpc.compression, e.g.
-            grpc.compression.Gzip. This is an EXPERIMENTAL option.
-
-        Returns:
-          A Call object instance which is an awaitable object.
-
-        Raises:
-          RpcError: Indicating 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.
-        """
         if compression:
             metadata = _compression.augment_metadata(metadata, compression)
 
@@ -277,11 +190,7 @@ class StreamStreamMultiCallable(_BaseMultiCallable):
         return call
 
 
-class Channel:
+class Channel(_base_channel.Channel):
-    """Asynchronous Channel implementation.
-
-    A cygrpc.AioChannel-backed implementation.
-    """
     _loop: asyncio.AbstractEventLoop
     _channel: cygrpc.AioChannel
     _unary_unary_interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]]
@@ -326,18 +235,9 @@ class Channel:
             self._loop)
 
     async def __aenter__(self):
-        """Starts an asynchronous context manager.
-
-        Returns:
-          Channel the channel that was instantiated.
-        """
         return self
 
     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.
-        """
         await self._close(None)
 
     async def _close(self, grace):
@@ -392,35 +292,10 @@ class Channel:
         self._channel.close()
 
     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.
-        """
         await self._close(grace)
 
     def get_state(self,
                   try_to_connect: bool = False) -> grpc.ChannelConnectivity:
-        """Check 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.
-        """
         result = self._channel.check_connectivity_state(try_to_connect)
         return _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[result]
 
@@ -428,31 +303,10 @@ class Channel:
             self,
             last_observed_state: grpc.ChannelConnectivity,
     ) -> None:
-        """Wait 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 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.
-        """
         assert await self._channel.watch_connectivity_state(
             last_observed_state.value[0], None)
 
     async def channel_ready(self) -> None:
-        """Creates a coroutine that ends when a Channel is ready."""
         state = self.get_state(try_to_connect=True)
         while state != grpc.ChannelConnectivity.READY:
             await self.wait_for_state_change(state)
@@ -464,19 +318,6 @@ class Channel:
             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 behaviour for serializing the request
-            message. Request goes unserialized in case None is passed.
-          response_deserializer: Optional behaviour for deserializing the
-            response message. Response goes undeserialized in case None
-            is passed.
-
-        Returns:
-          A UnaryUnaryMultiCallable value for the named unary-unary method.
-        """
         return UnaryUnaryMultiCallable(self._channel, _common.encode(method),
                                        request_serializer,
                                        response_deserializer,
@@ -513,3 +354,51 @@ class Channel:
                                          request_serializer,
                                          response_deserializer, None,
                                          self._loop)
+
+
+def insecure_channel(
+        target: str,
+        options: Optional[ChannelArgumentType] = None,
+        compression: Optional[grpc.Compression] = None,
+        interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]] = None):
+    """Creates an insecure asynchronous Channel to a server.
+
+    Args:
+      target: The server address
+      options: An optional list of key-value pairs (channel args
+        in gRPC Core runtime) to configure the channel.
+      compression: An optional value indicating the compression method to be
+        used over the lifetime of the channel. This is an EXPERIMENTAL option.
+      interceptors: An optional sequence of interceptors that will be executed for
+        any call executed with this channel.
+
+    Returns:
+      A Channel.
+    """
+    return Channel(target, () if options is None else options, None,
+                   compression, interceptors)
+
+
+def secure_channel(
+        target: str,
+        credentials: grpc.ChannelCredentials,
+        options: Optional[ChannelArgumentType] = None,
+        compression: Optional[grpc.Compression] = None,
+        interceptors: Optional[Sequence[UnaryUnaryClientInterceptor]] = None):
+    """Creates a secure asynchronous Channel to a server.
+
+    Args:
+      target: The server address.
+      credentials: A ChannelCredentials instance.
+      options: An optional list of key-value pairs (channel args
+        in gRPC Core runtime) to configure the channel.
+      compression: An optional value indicating the compression method to be
+        used over the lifetime of the channel. This is an EXPERIMENTAL option.
+      interceptors: An optional sequence of interceptors that will be executed for
+        any call executed with this channel.
+
+    Returns:
+      An aio.Channel.
+    """
+    return Channel(target, () if options is None else options,
+                   credentials._credentials, compression, interceptors)

src/python/grpcio/grpc/experimental/aio/_interceptor.py

@@ -35,6 +35,19 @@ class ClientCallDetails(
             'ClientCallDetails',
             ('method', 'timeout', 'metadata', 'credentials', 'wait_for_ready')),
         grpc.ClientCallDetails):
+    """Describes an RPC to be invoked.
+
+    This is an EXPERIMENTAL API.
+
+    Args:
+        method: The method name of the RPC.
+        timeout: An optional duration of time in seconds to allow for the RPC.
+        metadata: Optional metadata to be transmitted to the service-side of
+          the RPC.
+        credentials: An optional CallCredentials for the RPC.
+        wait_for_ready: This is an EXPERIMENTAL argument. An optional flag to
+          enable wait for ready mechanism.
+    """
 
     method: str
     timeout: Optional[float]

src/python/grpcio/grpc/experimental/aio/_server.py

@@ -21,6 +21,7 @@ import grpc
 from grpc import _common, _compression
 from grpc._cython import cygrpc
 
+from . import _base_server
 from ._typing import ChannelArgumentType
 
 
@@ -30,7 +31,7 @@ def _augment_channel_arguments(base_options: ChannelArgumentType,
     return tuple(base_options) + compression_option
 
 
-class Server:
+class Server(_base_server.Server):
     """Serves RPCs."""
 
     def __init__(self, thread_pool: Optional[Executor],