Merge pull request grpc#18732 from grpc/compression_reversion_reversion

Unrevert Python Compression

.pylintrc

@@ -6,6 +6,8 @@ ignore=
 	src/python/grpcio/grpc/framework/foundation,
 	src/python/grpcio/grpc/framework/interfaces,
 
+extension-pkg-whitelist=grpc._cython.cygrpc
+
 [VARIABLES]
 
 # TODO(https://github.com/PyCQA/pylint/issues/1345): How does the inspection
@@ -17,7 +19,7 @@ dummy-variables-rgx=^ignored_|^unused_
 # NOTE(nathaniel): Not particularly attached to this value; it just seems to
 # be what works for us at the moment (excepting the dead-code-walking Beta
 # API).
-max-args=6
+max-args=7
 
 [MISCELLANEOUS]
 

doc/python/sphinx/grpc.rst

@@ -172,3 +172,9 @@ Future Interfaces
 .. autoexception:: FutureTimeoutError
 .. autoexception:: FutureCancelledError
 .. autoclass:: Future
+
+
+Compression
+^^^^^^^^^^^
+
+.. autoclass:: Compression

src/python/grpcio/grpc/BUILD.bazel

@@ -12,6 +12,7 @@ py_library(
         ":channel",
         ":interceptor",
         ":server",
+        ":compression",
         "//src/python/grpcio/grpc/_cython:cygrpc",
         "//src/python/grpcio/grpc/experimental",
         "//src/python/grpcio/grpc/framework",
@@ -31,12 +32,18 @@ py_library(
     srcs = ["_auth.py"],
 )
 
+py_library(
+    name = "compression",
+    srcs = ["_compression.py"],
+)
+
 py_library(
     name = "channel",
     srcs = ["_channel.py"],
     deps = [
         ":common",
         ":grpcio_metadata",
+        ":compression",
     ],
 )
 
@@ -68,6 +75,7 @@ py_library(
     srcs = ["_server.py"],
     deps = [
         ":common",
+        ":compression",
         ":interceptor",
     ],
 )

src/python/grpcio/grpc/__init__.py

@@ -21,6 +21,7 @@
 import six
 
 from grpc._cython import cygrpc as _cygrpc
+from grpc import _compression
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())
 
@@ -413,6 +414,8 @@ class ClientCallDetails(six.with_metaclass(abc.ABCMeta)):
       credentials: An optional CallCredentials for the RPC.
       wait_for_ready: This is an EXPERIMENTAL argument. An optional flag t
         enable wait for ready mechanism.
+      compression: An element of grpc.compression, e.g.
+        grpc.compression.Gzip. This is an EXPERIMENTAL option.
     """
 
 
@@ -669,7 +672,8 @@ def __call__(self,
                  timeout=None,
                  metadata=None,
                  credentials=None,
-                 wait_for_ready=None):
+                 wait_for_ready=None,
+                 compression=None):
         """Synchronously invokes the underlying RPC.
 
         Args:
@@ -681,6 +685,8 @@ def __call__(self,
           credentials: An optional CallCredentials for the RPC.
           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:
           The response value for the RPC.
@@ -698,7 +704,8 @@ def with_call(self,
                   timeout=None,
                   metadata=None,
                   credentials=None,
-                  wait_for_ready=None):
+                  wait_for_ready=None,
+                  compression=None):
         """Synchronously invokes the underlying RPC.
 
         Args:
@@ -710,6 +717,8 @@ def with_call(self,
           credentials: An optional CallCredentials for the RPC.
           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:
           The response value for the RPC and a Call value for the RPC.
@@ -727,7 +736,8 @@ def future(self,
                timeout=None,
                metadata=None,
                credentials=None,
-               wait_for_ready=None):
+               wait_for_ready=None,
+               compression=None):
         """Asynchronously invokes the underlying RPC.
 
         Args:
@@ -739,6 +749,8 @@ def future(self,
           credentials: An optional CallCredentials for the RPC.
           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:
             An object that is both a Call for the RPC and a Future.
@@ -759,7 +771,8 @@ def __call__(self,
                  timeout=None,
                  metadata=None,
                  credentials=None,
-                 wait_for_ready=None):
+                 wait_for_ready=None,
+                 compression=None):
         """Invokes the underlying RPC.
 
         Args:
@@ -771,6 +784,8 @@ def __call__(self,
           credentials: An optional CallCredentials for the RPC.
           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:
             An object that is both a Call for the RPC and an iterator of
@@ -790,7 +805,8 @@ def __call__(self,
                  timeout=None,
                  metadata=None,
                  credentials=None,
-                 wait_for_ready=None):
+                 wait_for_ready=None,
+                 compression=None):
         """Synchronously invokes the underlying RPC.
 
         Args:
@@ -803,6 +819,8 @@ def __call__(self,
           credentials: An optional CallCredentials for the RPC.
           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:
           The response value for the RPC.
@@ -820,7 +838,8 @@ def with_call(self,
                   timeout=None,
                   metadata=None,
                   credentials=None,
-                  wait_for_ready=None):
+                  wait_for_ready=None,
+                  compression=None):
         """Synchronously invokes the underlying RPC on the client.
 
         Args:
@@ -833,6 +852,8 @@ def with_call(self,
           credentials: An optional CallCredentials for the RPC.
           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:
           The response value for the RPC and a Call object for the RPC.
@@ -850,7 +871,8 @@ def future(self,
                timeout=None,
                metadata=None,
                credentials=None,
-               wait_for_ready=None):
+               wait_for_ready=None,
+               compression=None):
         """Asynchronously invokes the underlying RPC on the client.
 
         Args:
@@ -862,6 +884,8 @@ def future(self,
           credentials: An optional CallCredentials for the RPC.
           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:
             An object that is both a Call for the RPC and a Future.
@@ -882,7 +906,8 @@ def __call__(self,
                  timeout=None,
                  metadata=None,
                  credentials=None,
-                 wait_for_ready=None):
+                 wait_for_ready=None,
+                 compression=None):
         """Invokes the underlying RPC on the client.
 
         Args:
@@ -894,6 +919,8 @@ def __call__(self,
           credentials: An optional CallCredentials for the RPC.
           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:
             An object that is both a Call for the RPC and an iterator of
@@ -1097,6 +1124,17 @@ def auth_context(self):
         """
         raise NotImplementedError()
 
+    def set_compression(self, compression):
+        """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.
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     def send_initial_metadata(self, initial_metadata):
         """Sends the initial metadata value to the client.
@@ -1184,6 +1222,16 @@ def set_details(self, details):
         """
         raise NotImplementedError()
 
+    def disable_next_message_compression(self):
+        """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.
+        """
+        raise NotImplementedError()
+
 
 #####################  Service-Side Handler Interfaces  ########################
 
@@ -1682,7 +1730,7 @@ def channel_ready_future(channel):
     return _utilities.channel_ready_future(channel)
 
 
-def insecure_channel(target, options=None):
+def insecure_channel(target, options=None, compression=None):
     """Creates an insecure Channel to a server.
 
     The returned Channel is thread-safe.
@@ -1691,15 +1739,18 @@ def insecure_channel(target, options=None):
       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.
 
     Returns:
       A Channel.
     """
     from grpc import _channel  # pylint: disable=cyclic-import
-    return _channel.Channel(target, () if options is None else options, None)
+    return _channel.Channel(target, ()
+                            if options is None else options, None, compression)
 
 
-def secure_channel(target, credentials, options=None):
+def secure_channel(target, credentials, options=None, compression=None):
     """Creates a secure Channel to a server.
 
     The returned Channel is thread-safe.
@@ -1709,13 +1760,15 @@ def secure_channel(target, credentials, options=None):
       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.
 
     Returns:
       A Channel.
     """
     from grpc import _channel  # pylint: disable=cyclic-import
     return _channel.Channel(target, () if options is None else options,
-                            credentials._credentials)
+                            credentials._credentials, compression)
 
 
 def intercept_channel(channel, *interceptors):
@@ -1750,7 +1803,8 @@ def server(thread_pool,
            handlers=None,
            interceptors=None,
            options=None,
-           maximum_concurrent_rpcs=None):
+           maximum_concurrent_rpcs=None,
+           compression=None):
     """Creates a Server with which RPCs can be serviced.
 
     Args:
@@ -1768,6 +1822,9 @@ def server(thread_pool,
       maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
         will service before returning RESOURCE_EXHAUSTED status, or None to
         indicate no limit.
+      compression: An element of grpc.compression, e.g.
+        grpc.compression.Gzip. This compression algorithm will be used for the
+        lifetime of the server unless overridden. This is an EXPERIMENTAL option.
 
     Returns:
       A Server object.
@@ -1777,7 +1834,7 @@ def server(thread_pool,
                                  if handlers is None else handlers, ()
                                  if interceptors is None else interceptors, ()
                                  if options is None else options,
-                                 maximum_concurrent_rpcs)
+                                 maximum_concurrent_rpcs, compression)
 
 
 @contextlib.contextmanager
@@ -1788,6 +1845,16 @@ def _create_servicer_context(rpc_event, state, request_deserializer):
     context._finalize_state()  # pylint: disable=protected-access
 
 
+class Compression(enum.IntEnum):
+    """Indicates the compression method to be used for an RPC.
+
+       This enumeration is part of an EXPERIMENTAL API.
+    """
+    NoCompression = _compression.NoCompression
+    Deflate = _compression.Deflate
+    Gzip = _compression.Gzip
+
+
 ###################################  __all__  #################################
 
 __all__ = (
@@ -1805,6 +1872,7 @@ def _create_servicer_context(rpc_event, state, request_deserializer):
     'AuthMetadataContext',
     'AuthMetadataPluginCallback',
     'AuthMetadataPlugin',
+    'Compression',
     'ClientCallDetails',
     'ServerCertificateConfiguration',
     'ServerCredentials',