diff --git a/apitools/base/py/compression.py b/apitools/base/py/compression.py
index 664f4386..327067e2 100644
--- a/apitools/base/py/compression.py
+++ b/apitools/base/py/compression.py
@@ -26,7 +26,8 @@
 
 
 # pylint: disable=invalid-name
-def CompressStream(in_stream, length, compresslevel=2, chunksize=16777216):
+def CompressStream(in_stream, length=None, compresslevel=2,
+                   chunksize=16777216):
 
     """Compresses an input stream into a file-like buffer.
 
@@ -43,7 +44,9 @@ def CompressStream(in_stream, length, compresslevel=2, chunksize=16777216):
             smaller than expected. Because data is written to the output
             buffer in increments of the chunksize, the output buffer may be
             larger than length by chunksize. Very uncompressible data can
-            exceed this further if gzip inflates the underlying data.
+            exceed this further if gzip inflates the underlying data. If
+            length is none, the input stream will be compressed until
+            it's exhausted.
         compresslevel: Optional, defaults to 2. The desired compression level.
         chunksize: Optional, defaults to 16MiB. The chunk size used when
             reading data from the input stream to write into the output
@@ -61,7 +64,7 @@ def CompressStream(in_stream, length, compresslevel=2, chunksize=16777216):
                        fileobj=out_stream,
                        compresslevel=compresslevel) as compress_stream:
         # Read until we've written at least length bytes to the output stream.
-        while out_stream.length < length:
+        while not length or out_stream.length < length:
             data = in_stream.read(chunksize)
             data_length = len(data)
             compress_stream.write(data)
diff --git a/apitools/base/py/compression_test.py b/apitools/base/py/compression_test.py
index 14c2a54e..c8ecdac1 100644
--- a/apitools/base/py/compression_test.py
+++ b/apitools/base/py/compression_test.py
@@ -50,6 +50,22 @@ def testCompressionExhausted(self):
         # Ensure the input stream was exhausted.
         self.assertTrue(exhausted)
 
+    def testCompressionUnbounded(self):
+        """Test unbounded compression.
+
+        Test that the input stream is exhausted when length is none.
+        """
+        output, read, exhausted = compression.CompressStream(
+            self.stream,
+            None,
+            9)
+        # Ensure the compressed buffer is smaller than the input buffer.
+        self.assertLess(output.length, self.length)
+        # Ensure we read the entire input stream.
+        self.assertEqual(read, self.length)
+        # Ensure the input stream was exhausted.
+        self.assertTrue(exhausted)
+
     def testCompressionPartial(self):
         """Test partial compression.
 
diff --git a/apitools/base/py/gzip.py b/apitools/base/py/gzip.py
index 1fddb8bf..e9eed144 100644
--- a/apitools/base/py/gzip.py
+++ b/apitools/base/py/gzip.py
@@ -47,7 +47,7 @@ def write32u(output, value):
     output.write(struct.pack("<L", value))
 
 
-class _PaddedFile:
+class _PaddedFile(object):
     """Minimal read-only file object that prepends a string to the contents
     of an actual file. Shouldn't be used outside of gzip.py, as it lacks
     essential functionality."""
diff --git a/apitools/base/py/transfer.py b/apitools/base/py/transfer.py
index e6b5df63..e795a1a2 100644
--- a/apitools/base/py/transfer.py
+++ b/apitools/base/py/transfer.py
@@ -549,7 +549,7 @@ class Upload(_Transfer):
     def __init__(self, stream, mime_type, total_size=None, http=None,
                  close_stream=False, chunksize=None, auto_transfer=True,
                  progress_callback=None, finish_callback=None,
-                 **kwds):
+                 gzip_encoded=False, **kwds):
         super(Upload, self).__init__(
             stream, close_stream=close_stream, chunksize=chunksize,
             auto_transfer=auto_transfer, http=http, **kwds)
@@ -560,6 +560,7 @@ def __init__(self, stream, mime_type, total_size=None, http=None,
         self.__server_chunk_granularity = None
         self.__strategy = None
         self.__total_size = None
+        self.__gzip_encoded = gzip_encoded
 
         self.progress_callback = progress_callback
         self.finish_callback = finish_callback
@@ -570,7 +571,8 @@ def progress(self):
         return self.__progress
 
     @classmethod
-    def FromFile(cls, filename, mime_type=None, auto_transfer=True, **kwds):
+    def FromFile(cls, filename, mime_type=None, auto_transfer=True,
+                 gzip_encoded=False, **kwds):
         """Create a new Upload object from a filename."""
         path = os.path.expanduser(filename)
         if not os.path.exists(path):
@@ -582,20 +584,23 @@ def FromFile(cls, filename, mime_type=None, auto_transfer=True, **kwds):
                     'Could not guess mime type for %s' % path)
         size = os.stat(path).st_size
         return cls(open(path, 'rb'), mime_type, total_size=size,
-                   close_stream=True, auto_transfer=auto_transfer, **kwds)
+                   close_stream=True, auto_transfer=auto_transfer,
+                   gzip_encoded=gzip_encoded, **kwds)
 
     @classmethod
     def FromStream(cls, stream, mime_type, total_size=None, auto_transfer=True,
-                   **kwds):
+                   gzip_encoded=False, **kwds):
         """Create a new Upload object from a stream."""
         if mime_type is None:
             raise exceptions.InvalidUserInputError(
                 'No mime_type specified for stream')
         return cls(stream, mime_type, total_size=total_size,
-                   close_stream=False, auto_transfer=auto_transfer, **kwds)
+                   close_stream=False, auto_transfer=auto_transfer,
+                   gzip_encoded=gzip_encoded, **kwds)
 
     @classmethod
-    def FromData(cls, stream, json_data, http, auto_transfer=None, **kwds):
+    def FromData(cls, stream, json_data, http, auto_transfer=None,
+                 gzip_encoded=False, **kwds):
         """Create a new Upload of stream from serialized json_data and http."""
         info = json.loads(json_data)
         missing_keys = cls._REQUIRED_SERIALIZATION_KEYS - set(info.keys())
@@ -607,7 +612,8 @@ def FromData(cls, stream, json_data, http, auto_transfer=None, **kwds):
             raise exceptions.InvalidUserInputError(
                 'Cannot override total_size on serialized Upload')
         upload = cls.FromStream(stream, info['mime_type'],
-                                total_size=info.get('total_size'), **kwds)
+                                total_size=info.get('total_size'),
+                                gzip_encoded=gzip_encoded, **kwds)
         if isinstance(stream, io.IOBase) and not stream.seekable():
             raise exceptions.InvalidUserInputError(
                 'Cannot restart resumable upload on non-seekable stream')
@@ -724,6 +730,17 @@ def ConfigureRequest(self, upload_config, http_request, url_builder):
             else:
                 url_builder.query_params['uploadType'] = 'media'
                 self.__ConfigureMediaRequest(http_request)
+            # Once the entire body is written, compress the body if configured
+            # to. Both multipart and media request uploads will read the
+            # entire stream into memory, which means full compression is also
+            # safe to perform. Because the strategy is set to SIMPLE_UPLOAD,
+            # StreamInChunks throws an exception, meaning double compression
+            # cannot happen.
+            if self.__gzip_encoded:
+                http_request.headers['Content-Encoding'] = 'gzip'
+                body_buffer = six.BytesIO(http_request.body)
+                body, _, _ = compression.CompressStream(body_buffer)
+                http_request.body = body
         else:
             url_builder.relative_path = upload_config.resumable_path
             url_builder.query_params['uploadType'] = 'resumable'
@@ -866,8 +883,7 @@ def __ValidateChunksize(self, chunksize=None):
                 self.__server_chunk_granularity)
 
     def __StreamMedia(self, callback=None, finish_callback=None,
-                      additional_headers=None, use_chunks=True,
-                      compressed=False):
+                      additional_headers=None, use_chunks=True):
         """Helper function for StreamMedia / StreamInChunks."""
         if self.strategy != RESUMABLE_UPLOAD:
             raise exceptions.InvalidUserInputError(
@@ -879,14 +895,16 @@ def __StreamMedia(self, callback=None, finish_callback=None,
 
         def CallSendChunk(start):
             return self.__SendChunk(
-                start, additional_headers=additional_headers,
-                compressed=compressed)
+                start, additional_headers=additional_headers)
 
         def CallSendMediaBody(start):
             return self.__SendMediaBody(
                 start, additional_headers=additional_headers)
 
         send_func = CallSendChunk if use_chunks else CallSendMediaBody
+        if not use_chunks and self.__gzip_encoded:
+            raise exceptions.InvalidUserInputError(
+                'Cannot gzip encode non-chunked upload')
         if use_chunks:
             self.__ValidateChunksize(self.chunksize)
         self.EnsureInitialized()
@@ -934,11 +952,11 @@ def StreamMedia(self, callback=None, finish_callback=None,
             additional_headers=additional_headers, use_chunks=False)
 
     def StreamInChunks(self, callback=None, finish_callback=None,
-                       additional_headers=None, compressed=False):
+                       additional_headers=None):
         """Send this (resumable) upload in chunks."""
         return self.__StreamMedia(
             callback=callback, finish_callback=finish_callback,
-            additional_headers=additional_headers, compressed=compressed)
+            additional_headers=additional_headers)
 
     def __SendMediaRequest(self, request, end):
         """Request helper function for SendMediaBody & SendChunk."""
@@ -983,12 +1001,12 @@ def __SendMediaBody(self, start, additional_headers=None):
 
         return self.__SendMediaRequest(request, self.total_size)
 
-    def __SendChunk(self, start, additional_headers=None, compressed=False):
+    def __SendChunk(self, start, additional_headers=None):
         """Send the specified chunk."""
         self.EnsureInitialized()
         no_log_body = self.total_size is None
         request = http_wrapper.Request(url=self.url, http_method='PUT')
-        if compressed:
+        if self.__gzip_encoded:
             request.headers['Content-Encoding'] = 'gzip'
             body_stream, read_length, exhausted = compression.CompressStream(
                 self.stream, self.chunksize)
diff --git a/apitools/base/py/transfer_test.py b/apitools/base/py/transfer_test.py
index 2b22caa5..1febbd49 100644
--- a/apitools/base/py/transfer_test.py
+++ b/apitools/base/py/transfer_test.py
@@ -23,6 +23,8 @@
 import unittest2
 
 from apitools.base.py import base_api
+from apitools.base.py import exceptions
+from apitools.base.py import gzip
 from apitools.base.py import http_wrapper
 from apitools.base.py import transfer
 
@@ -307,41 +309,171 @@ def testMultipartEncoding(self):
             rewritten_upload_contents = http_request.body
             self.assertTrue(rewritten_upload_contents.endswith(upload_bytes))
 
-    def testCompressedChunkedUpload(self):
-        """Test that the compression flag is propagated and runs."""
 
+class CompressedUploadTest(unittest2.TestCase):
+
+    def setUp(self):
         # Sample highly compressible data.
-        sample_data = b'abc' * 100
-        sample_response = http_wrapper.Response(
-            info={'status': http_client.OK},
+        self.sample_data = b'abc' * 200
+        # Stream of the sample data.
+        self.sample_stream = six.BytesIO(self.sample_data)
+        # Sample url_builder.
+        self.url_builder = base_api._UrlBuilder('http://www.uploads.com')
+        # Sample request.
+        self.request = http_wrapper.Request(
+            'http://www.uploads.com',
+            headers={'content-type': 'text/plain'})
+        # Sample successful response.
+        self.response = http_wrapper.Response(
+            info={'status': http_client.OK,
+                  'location': 'http://www.uploads.com'},
             content='',
-            request_url='url',
-        )
-
+            request_url='http://www.uploads.com',)
+
+    def testStreamInChunksCompressed(self):
+        """Test that StreamInChunks will handle compression correctly."""
+        # Create and configure the upload object.
+        upload = transfer.Upload(
+            stream=self.sample_stream,
+            mime_type='text/plain',
+            total_size=len(self.sample_data),
+            close_stream=False,
+            gzip_encoded=True)
+        upload.strategy = transfer.RESUMABLE_UPLOAD
+        # Set the chunk size so the entire stream is uploaded.
+        upload.chunksize = len(self.sample_data)
         # Mock the upload to return the sample response.
         with mock.patch.object(transfer.Upload,
-                               '_Upload__SendMediaRequest') as mock_result:
-            mock_result.return_value = sample_response
-
-            upload = transfer.Upload.FromStream(
-                six.BytesIO(sample_data),
-                'text/plain',
-                total_size=len(sample_data))
-            upload.strategy = transfer.RESUMABLE_UPLOAD
-            # Set the chunk size so the entire stream is uploaded.
-            upload.chunksize = len(sample_data)
-            # Skip actual initialization.
-            upload._Initialize('http', 'url')
-            upload.StreamInChunks(compressed=True)
-
+                               '_Upload__SendMediaRequest') as mock_result, \
+                mock.patch.object(http_wrapper,
+                                  'MakeRequest') as make_request:
+            mock_result.return_value = self.response
+            make_request.return_value = self.response
+
+            # Initialization.
+            upload.InitializeUpload(self.request, 'http')
+            upload.StreamInChunks()
             # Get the uploaded request and end position of the stream.
             (request, end), _ = mock_result.call_args_list[0]
-
             # Ensure the mock was called.
             self.assertTrue(mock_result.called)
             # Ensure the stream was fully read.
-            self.assertEqual(len(sample_data), end)
+            self.assertEqual(len(self.sample_data), end)
             # Ensure the correct content encoding was set.
             self.assertEqual(request.headers['Content-Encoding'], 'gzip')
             # Ensure the stream was compresed.
-            self.assertLess(len(request.body), len(sample_data))
+            self.assertLess(len(request.body), len(self.sample_data))
+
+    def testStreamMediaCompressedFail(self):
+        """Test that non-chunked uploads raise an exception.
+
+        Ensure uploads with the compressed and resumable flags set called from
+        StreamMedia raise an exception. Those uploads are unsupported.
+        """
+        # Create the upload object.
+        upload = transfer.Upload(
+            stream=self.sample_stream,
+            mime_type='text/plain',
+            total_size=len(self.sample_data),
+            close_stream=False,
+            auto_transfer=True,
+            gzip_encoded=True)
+        upload.strategy = transfer.RESUMABLE_UPLOAD
+        # Mock the upload to return the sample response.
+        with mock.patch.object(http_wrapper,
+                               'MakeRequest') as make_request:
+            make_request.return_value = self.response
+
+            # Initialization.
+            upload.InitializeUpload(self.request, 'http')
+            # Ensure stream media raises an exception when the upload is
+            # compressed. Compression is not supported on non-chunked uploads.
+            with self.assertRaises(exceptions.InvalidUserInputError):
+                upload.StreamMedia()
+
+    def testAutoTransferCompressed(self):
+        """Test that automatic transfers are compressed.
+
+        Ensure uploads with the compressed, resumable, and automatic transfer
+        flags set call StreamInChunks. StreamInChunks is tested in an earlier
+        test.
+        """
+        # Create the upload object.
+        upload = transfer.Upload(
+            stream=self.sample_stream,
+            mime_type='text/plain',
+            total_size=len(self.sample_data),
+            close_stream=False,
+            gzip_encoded=True)
+        upload.strategy = transfer.RESUMABLE_UPLOAD
+        # Mock the upload to return the sample response.
+        with mock.patch.object(transfer.Upload,
+                               'StreamInChunks') as mock_result, \
+                mock.patch.object(http_wrapper,
+                                  'MakeRequest') as make_request:
+            mock_result.return_value = self.response
+            make_request.return_value = self.response
+
+            # Initialization.
+            upload.InitializeUpload(self.request, 'http')
+            # Ensure the mock was called.
+            self.assertTrue(mock_result.called)
+
+    def testMultipartCompressed(self):
+        """Test that multipart uploads are compressed."""
+        # Create the multipart configuration.
+        upload_config = base_api.ApiUploadInfo(
+            accept=['*/*'],
+            max_size=None,
+            simple_multipart=True,
+            simple_path=u'/upload',)
+        # Create the upload object.
+        upload = transfer.Upload(
+            stream=self.sample_stream,
+            mime_type='text/plain',
+            total_size=len(self.sample_data),
+            close_stream=False,
+            gzip_encoded=True)
+        # Set a body to trigger multipart configuration.
+        self.request.body = '{"body_field_one": 7}'
+        # Configure the request.
+        upload.ConfigureRequest(upload_config, self.request, self.url_builder)
+        # Ensure the request is a multipart request now.
+        self.assertEqual(
+            self.url_builder.query_params['uploadType'], 'multipart')
+        # Ensure the request is gzip encoded.
+        self.assertEqual(self.request.headers['Content-Encoding'], 'gzip')
+        # Ensure data is compressed
+        self.assertLess(len(self.request.body), len(self.sample_data))
+        # Ensure uncompressed data includes the sample data.
+        with gzip.GzipFile(fileobj=self.request.body) as f:
+            original = f.read()
+            self.assertTrue(self.sample_data in original)
+
+    def testMediaCompressed(self):
+        """Test that media uploads are compressed."""
+        # Create the media configuration.
+        upload_config = base_api.ApiUploadInfo(
+            accept=['*/*'],
+            max_size=None,
+            simple_multipart=True,
+            simple_path=u'/upload',)
+        # Create the upload object.
+        upload = transfer.Upload(
+            stream=self.sample_stream,
+            mime_type='text/plain',
+            total_size=len(self.sample_data),
+            close_stream=False,
+            gzip_encoded=True)
+        # Configure the request.
+        upload.ConfigureRequest(upload_config, self.request, self.url_builder)
+        # Ensure the request is a media request now.
+        self.assertEqual(self.url_builder.query_params['uploadType'], 'media')
+        # Ensure the request is gzip encoded.
+        self.assertEqual(self.request.headers['Content-Encoding'], 'gzip')
+        # Ensure data is compressed
+        self.assertLess(len(self.request.body), len(self.sample_data))
+        # Ensure uncompressed data includes the sample data.
+        with gzip.GzipFile(fileobj=self.request.body) as f:
+            original = f.read()
+            self.assertTrue(self.sample_data in original)