diff --git a/apitools/base/py/compression.py b/apitools/base/py/compression.py
new file mode 100644
index 00000000..664f4386
--- /dev/null
+++ b/apitools/base/py/compression.py
@@ -0,0 +1,135 @@
+#!/usr/bin/env python
+#
+# Copyright 2017 Google Inc.
+#
+# 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.
+
+"""Compression support for apitools."""
+
+from collections import deque
+
+from apitools.base.py import gzip
+
+__all__ = [
+    'CompressStream',
+]
+
+
+# pylint: disable=invalid-name
+def CompressStream(in_stream, length, compresslevel=2, chunksize=16777216):
+
+    """Compresses an input stream into a file-like buffer.
+
+    This reads from the input stream until either we've stored at least length
+    compressed bytes, or the input stream has been exhausted.
+
+    This supports streams of unknown size.
+
+    Args:
+        in_stream: The input stream to read from.
+        length: The target number of compressed bytes to buffer in the output
+            stream. The actual length of the output buffer can vary from this
+            length. If the input stream is exhaused, the output buffer may be
+            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.
+        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
+            buffer.
+
+    Returns:
+        A file-like output buffer of compressed bytes, the number of bytes read
+        from the input stream, and a flag denoting if the input stream was
+        exhausted.
+    """
+    in_read = 0
+    in_exhausted = False
+    out_stream = StreamingBuffer()
+    with gzip.GzipFile(mode='wb',
+                       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:
+            data = in_stream.read(chunksize)
+            data_length = len(data)
+            compress_stream.write(data)
+            in_read += data_length
+            # If we read less than requested, the stream is exhausted.
+            if data_length < chunksize:
+                in_exhausted = True
+                break
+    return out_stream, in_read, in_exhausted
+
+
+class StreamingBuffer(object):
+
+    """Provides a file-like object that writes to a temporary buffer.
+
+    When data is read from the buffer, it is permanently removed. This is
+    useful when there are memory constraints preventing the entire buffer from
+    being stored in memory.
+    """
+
+    def __init__(self):
+        # The buffer of byte arrays.
+        self.__buf = deque()
+        # The number of bytes in __buf.
+        self.__size = 0
+
+    def __len__(self):
+        return self.__size
+
+    def __nonzero__(self):
+        # For 32-bit python2.x, len() cannot exceed a 32-bit number; avoid
+        # accidental len() calls from httplib in the form of "if this_object:".
+        return bool(self.__size)
+
+    @property
+    def length(self):
+        # For 32-bit python2.x, len() cannot exceed a 32-bit number.
+        return self.__size
+
+    def write(self, data):
+        # Gzip can write many 0 byte chunks for highly compressible data.
+        # Prevent them from being added internally.
+        if data is not None and data:
+            self.__buf.append(data)
+            self.__size += len(data)
+
+    def read(self, size=None):
+        """Read at most size bytes from this buffer.
+
+        Bytes read from this buffer are consumed and are permanently removed.
+
+        Args:
+          size: If provided, read no more than size bytes from the buffer.
+            Otherwise, this reads the entire buffer.
+
+        Returns:
+          The bytes read from this buffer.
+        """
+        if size is None:
+            size = self.__size
+        ret_list = []
+        while size > 0 and self.__buf:
+            data = self.__buf.popleft()
+            size -= len(data)
+            ret_list.append(data)
+        if size < 0:
+            ret_list[-1], remainder = ret_list[-1][:size], ret_list[-1][size:]
+            self.__buf.appendleft(remainder)
+        ret = b''.join(ret_list)
+        self.__size -= len(ret)
+        return ret
diff --git a/apitools/base/py/compression_test.py b/apitools/base/py/compression_test.py
new file mode 100644
index 00000000..14c2a54e
--- /dev/null
+++ b/apitools/base/py/compression_test.py
@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+#
+# Copyright 2017 Google Inc.
+#
+# 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.
+
+"""Tests for compression."""
+
+from apitools.base.py import compression
+from apitools.base.py import gzip
+
+import six
+import unittest2
+
+
+class CompressionTest(unittest2.TestCase):
+
+    def setUp(self):
+        # Sample highly compressible data (~50MB).
+        self.sample_data = b'abc' * 16777216
+        # Stream of the sample data.
+        self.stream = six.BytesIO()
+        self.stream.write(self.sample_data)
+        self.length = self.stream.tell()
+        self.stream.seek(0)
+
+    def testCompressionExhausted(self):
+        """Test full compression.
+
+        Test that highly compressible data is actually compressed in entirety.
+        """
+        output, read, exhausted = compression.CompressStream(
+            self.stream,
+            self.length,
+            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.
+
+        Test that the length parameter works correctly. The amount of data
+        that's compressed can be greater than or equal to the requested length.
+        """
+        output_length = 40
+        output, _, exhausted = compression.CompressStream(
+            self.stream,
+            output_length,
+            9)
+        # Ensure the requested read size is <= the compressed buffer size.
+        self.assertLessEqual(output_length, output.length)
+        # Ensure the input stream was not exhausted.
+        self.assertFalse(exhausted)
+
+    def testCompressionIntegrity(self):
+        """Test that compressed data can be decompressed."""
+        output, read, exhausted = compression.CompressStream(
+            self.stream,
+            self.length,
+            9)
+        # Ensure uncompressed data matches the sample data.
+        with gzip.GzipFile(fileobj=output) as f:
+            original = f.read()
+            self.assertEqual(original, self.sample_data)
+        # Ensure we read the entire input stream.
+        self.assertEqual(read, self.length)
+        # Ensure the input stream was exhausted.
+        self.assertTrue(exhausted)
+
+
+class StreamingBufferTest(unittest2.TestCase):
+
+    def setUp(self):
+        self.stream = compression.StreamingBuffer()
+
+    def testSimpleStream(self):
+        """Test simple stream operations.
+
+        Test that the stream can be written to and read from. Also test that
+        reading from the stream consumes the bytes.
+        """
+        # Ensure the stream is empty.
+        self.assertEqual(self.stream.length, 0)
+        # Ensure data is correctly written.
+        self.stream.write(b'Sample data')
+        self.assertEqual(self.stream.length, 11)
+        # Ensure data can be read and the read data is purged from the stream.
+        data = self.stream.read(11)
+        self.assertEqual(data, b'Sample data')
+        self.assertEqual(self.stream.length, 0)
+
+    def testPartialReads(self):
+        """Test partial stream reads.
+
+        Test that the stream can be read in chunks while perserving the
+        consumption mechanics.
+        """
+        self.stream.write(b'Sample data')
+        # Ensure data can be read and the read data is purged from the stream.
+        data = self.stream.read(6)
+        self.assertEqual(data, b'Sample')
+        self.assertEqual(self.stream.length, 5)
+        # Ensure the remaining data can be read.
+        data = self.stream.read(5)
+        self.assertEqual(data, b' data')
+        self.assertEqual(self.stream.length, 0)
+
+    def testTooShort(self):
+        """Test excessive stream reads.
+
+        Test that more data can be requested from the stream than available
+        without raising an exception.
+        """
+        self.stream.write(b'Sample')
+        # Ensure requesting more data than available does not raise an
+        # exception.
+        data = self.stream.read(100)
+        self.assertEqual(data, b'Sample')
+        self.assertEqual(self.stream.length, 0)
diff --git a/apitools/base/py/gzip.py b/apitools/base/py/gzip.py
new file mode 100644
index 00000000..1fddb8bf
--- /dev/null
+++ b/apitools/base/py/gzip.py
@@ -0,0 +1,617 @@
+# Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+# 2011, 2012, 2013, 2014, 2015, 2016, 2017 Python Software Foundation; All
+# Rights Reserved
+#
+# This is a backport from python 3.4 into python 2.7. Text and exclusive mode
+# support are removed as they're unsupported in 2.7. This backport patches a
+# streaming bug that exists in python 2.7.
+
+"""Functions that read and write gzipped files.
+
+The user of the file doesn't have to worry about the compression,
+but random access is not allowed."""
+
+# based on Andrew Kuchling's minigzip.py distributed with the zlib module
+
+import six
+from six.moves import builtins
+from six.moves import range
+
+import struct
+import sys
+import time
+import os
+import zlib
+import io
+
+__all__ = ["GzipFile", "open", "compress", "decompress"]
+
+FTEXT, FHCRC, FEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16
+
+READ, WRITE = 1, 2
+
+
+def open(filename, mode="rb", compresslevel=9):
+    """Shorthand for GzipFile(filename, mode, compresslevel).
+
+    The filename argument is required; mode defaults to 'rb'
+    and compresslevel defaults to 9.
+
+    """
+    return GzipFile(filename, mode, compresslevel)
+
+
+def write32u(output, value):
+    # The L format writes the bit pattern correctly whether signed
+    # or unsigned.
+    output.write(struct.pack("<L", value))
+
+
+class _PaddedFile:
+    """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."""
+
+    def __init__(self, f, prepend=b''):
+        self._buffer = prepend
+        self._length = len(prepend)
+        self.file = f
+        self._read = 0
+
+    def read(self, size):
+        if self._read is None:
+            return self.file.read(size)
+        if self._read + size <= self._length:
+            read = self._read
+            self._read += size
+            return self._buffer[read:self._read]
+        else:
+            read = self._read
+            self._read = None
+            return self._buffer[read:] + \
+                self.file.read(size - self._length + read)
+
+    def prepend(self, prepend=b'', readprevious=False):
+        if self._read is None:
+            self._buffer = prepend
+        elif readprevious and len(prepend) <= self._read:
+            self._read -= len(prepend)
+            return
+        else:
+            self._buffer = self._buffer[self._read:] + prepend
+        self._length = len(self._buffer)
+        self._read = 0
+
+    def unused(self):
+        if self._read is None:
+            return b''
+        return self._buffer[self._read:]
+
+    def seek(self, offset, whence=0):
+        # This is only ever called with offset=whence=0
+        if whence == 1 and self._read is not None:
+            if 0 <= offset + self._read <= self._length:
+                self._read += offset
+                return
+            else:
+                offset += self._length - self._read
+        self._read = None
+        self._buffer = None
+        return self.file.seek(offset, whence)
+
+    def __getattr__(self, name):
+        return getattr(self.file, name)
+
+
+class GzipFile(io.BufferedIOBase):
+    """The GzipFile class simulates most of the methods of a file object with
+    the exception of the readinto() and truncate() methods.
+
+    This class only supports opening files in binary mode. If you need to open
+    a compressed file in text mode, use the gzip.open() function.
+
+    """
+
+    myfileobj = None
+    max_read_chunk = 10 * 1024 * 1024   # 10Mb
+
+    def __init__(self, filename=None, mode=None,
+                 compresslevel=9, fileobj=None, mtime=None):
+        """Constructor for the GzipFile class.
+
+        At least one of fileobj and filename must be given a
+        non-trivial value.
+
+        The new class instance is based on fileobj, which can be a regular
+        file, an io.BytesIO object, or any other object which simulates a file.
+        It defaults to None, in which case filename is opened to provide
+        a file object.
+
+        When fileobj is not None, the filename argument is only used to be
+        included in the gzip file header, which may includes the original
+        filename of the uncompressed file.  It defaults to the filename of
+        fileobj, if discernible; otherwise, it defaults to the empty string,
+        and in this case the original filename is not included in the header.
+
+        The mode argument can be any of 'r', 'rb', 'a', 'ab', 'w', or 'wb',
+        depending on whether the file will be read or written.  The default
+        is the mode of fileobj if discernible; otherwise, the default is 'rb'.
+        A mode of 'r' is equivalent to one of 'rb', and similarly for 'w' and
+        'wb', and 'a' and 'ab'.
+
+        The compresslevel argument is an integer from 0 to 9 controlling the
+        level of compression; 1 is fastest and produces the least compression,
+        and 9 is slowest and produces the most compression. 0 is no compression
+        at all. The default is 9.
+
+        The mtime argument is an optional numeric timestamp to be written
+        to the stream when compressing.  All gzip compressed streams
+        are required to contain a timestamp.  If omitted or None, the
+        current time is used.  This module ignores the timestamp when
+        decompressing; however, some programs, such as gunzip, make use
+        of it.  The format of the timestamp is the same as that of the
+        return value of time.time() and of the st_mtime member of the
+        object returned by os.stat().
+
+        """
+
+        if mode and ('t' in mode or 'U' in mode):
+            raise ValueError("Invalid mode: {!r}".format(mode))
+        if mode and 'b' not in mode:
+            mode += 'b'
+        if fileobj is None:
+            fileobj = self.myfileobj = builtins.open(filename, mode or 'rb')
+        if filename is None:
+            filename = getattr(fileobj, 'name', '')
+            if not isinstance(filename, six.string_types):
+                filename = ''
+        if mode is None:
+            mode = getattr(fileobj, 'mode', 'rb')
+
+        if mode.startswith('r'):
+            self.mode = READ
+            # Set flag indicating start of a new member
+            self._new_member = True
+            # Buffer data read from gzip file. extrastart is offset in
+            # stream where buffer starts. extrasize is number of
+            # bytes remaining in buffer from current stream position.
+            self.extrabuf = b""
+            self.extrasize = 0
+            self.extrastart = 0
+            self.name = filename
+            # Starts small, scales exponentially
+            self.min_readsize = 100
+            fileobj = _PaddedFile(fileobj)
+
+        elif mode.startswith(('w', 'a')):
+            self.mode = WRITE
+            self._init_write(filename)
+            self.compress = zlib.compressobj(compresslevel,
+                                             zlib.DEFLATED,
+                                             -zlib.MAX_WBITS,
+                                             zlib.DEF_MEM_LEVEL,
+                                             0)
+        else:
+            raise ValueError("Invalid mode: {!r}".format(mode))
+
+        self.fileobj = fileobj
+        self.offset = 0
+        self.mtime = mtime
+
+        if self.mode == WRITE:
+            self._write_gzip_header()
+
+    @property
+    def filename(self):
+        import warnings
+        warnings.warn("use the name attribute", DeprecationWarning, 2)
+        if self.mode == WRITE and self.name[-3:] != ".gz":
+            return self.name + ".gz"
+        return self.name
+
+    def __repr__(self):
+        fileobj = self.fileobj
+        if isinstance(fileobj, _PaddedFile):
+            fileobj = fileobj.file
+        s = repr(fileobj)
+        return '<gzip ' + s[1:-1] + ' ' + hex(id(self)) + '>'
+
+    def _check_closed(self):
+        """Raises a ValueError if the underlying file object has been closed.
+
+        """
+        if self.closed:
+            raise ValueError('I/O operation on closed file.')
+
+    def _init_write(self, filename):
+        self.name = filename
+        self.crc = zlib.crc32(b"") & 0xffffffff
+        self.size = 0
+        self.writebuf = []
+        self.bufsize = 0
+
+    def _write_gzip_header(self):
+        self.fileobj.write(b'\037\213')             # magic header
+        self.fileobj.write(b'\010')                 # compression method
+        try:
+            # RFC 1952 requires the FNAME field to be Latin-1. Do not
+            # include filenames that cannot be represented that way.
+            fname = os.path.basename(self.name)
+            if not isinstance(fname, six.binary_type):
+                fname = fname.encode('latin-1')
+            if fname.endswith(b'.gz'):
+                fname = fname[:-3]
+        except UnicodeEncodeError:
+            fname = b''
+        flags = 0
+        if fname:
+            flags = FNAME
+        self.fileobj.write(six.unichr(flags).encode('latin-1'))
+        mtime = self.mtime
+        if mtime is None:
+            mtime = time.time()
+        write32u(self.fileobj, int(mtime))
+        self.fileobj.write(b'\002')
+        self.fileobj.write(b'\377')
+        if fname:
+            self.fileobj.write(fname + b'\000')
+
+    def _init_read(self):
+        self.crc = zlib.crc32(b"") & 0xffffffff
+        self.size = 0
+
+    def _read_exact(self, n):
+        data = self.fileobj.read(n)
+        while len(data) < n:
+            b = self.fileobj.read(n - len(data))
+            if not b:
+                raise EOFError("Compressed file ended before the "
+                               "end-of-stream marker was reached")
+            data += b
+        return data
+
+    def _read_gzip_header(self):
+        magic = self.fileobj.read(2)
+        if magic == b'':
+            return False
+
+        if magic != b'\037\213':
+            raise OSError('Not a gzipped file')
+
+        method, flag, self.mtime = struct.unpack("<BBIxx", self._read_exact(8))
+        if method != 8:
+            raise OSError('Unknown compression method')
+
+        if flag & FEXTRA:
+            # Read & discard the extra field, if present
+            extra_len, = struct.unpack("<H", self._read_exact(2))
+            self._read_exact(extra_len)
+        if flag & FNAME:
+            # Read and discard a null-terminated string containing the filename
+            while True:
+                s = self.fileobj.read(1)
+                if not s or s == b'\000':
+                    break
+        if flag & FCOMMENT:
+            # Read and discard a null-terminated string containing a comment
+            while True:
+                s = self.fileobj.read(1)
+                if not s or s == b'\000':
+                    break
+        if flag & FHCRC:
+            self._read_exact(2)     # Read & discard the 16-bit header CRC
+
+        unused = self.fileobj.unused()
+        if unused:
+            uncompress = self.decompress.decompress(unused)
+            self._add_read_data(uncompress)
+        return True
+
+    def write(self, data):
+        self._check_closed()
+        if self.mode != WRITE:
+            import errno
+            raise OSError(errno.EBADF, "write() on read-only GzipFile object")
+
+        if self.fileobj is None:
+            raise ValueError("write() on closed GzipFile object")
+
+        # Convert data type if called by io.BufferedWriter.
+        if isinstance(data, memoryview):
+            data = data.tobytes()
+
+        if len(data) > 0:
+            self.fileobj.write(self.compress.compress(data))
+            self.size += len(data)
+            self.crc = zlib.crc32(data, self.crc) & 0xffffffff
+            self.offset += len(data)
+
+        return len(data)
+
+    def read(self, size=-1):
+        self._check_closed()
+        if self.mode != READ:
+            import errno
+            raise OSError(errno.EBADF, "read() on write-only GzipFile object")
+
+        if self.extrasize <= 0 and self.fileobj is None:
+            return b''
+
+        readsize = 1024
+        if size < 0:        # get the whole thing
+            while self._read(readsize):
+                readsize = min(self.max_read_chunk, readsize * 2)
+            size = self.extrasize
+        else:               # just get some more of it
+            while size > self.extrasize:
+                if not self._read(readsize):
+                    if size > self.extrasize:
+                        size = self.extrasize
+                    break
+                readsize = min(self.max_read_chunk, readsize * 2)
+
+        offset = self.offset - self.extrastart
+        chunk = self.extrabuf[offset: offset + size]
+        self.extrasize = self.extrasize - size
+
+        self.offset += size
+        return chunk
+
+    def read1(self, size=-1):
+        self._check_closed()
+        if self.mode != READ:
+            import errno
+            raise OSError(errno.EBADF, "read1() on write-only GzipFile object")
+
+        if self.extrasize <= 0 and self.fileobj is None:
+            return b''
+
+        # For certain input data, a single call to _read() may not return
+        # any data. In this case, retry until we get some data or reach EOF.
+        while self.extrasize <= 0 and self._read():
+            pass
+        if size < 0 or size > self.extrasize:
+            size = self.extrasize
+
+        offset = self.offset - self.extrastart
+        chunk = self.extrabuf[offset: offset + size]
+        self.extrasize -= size
+        self.offset += size
+        return chunk
+
+    def peek(self, n):
+        if self.mode != READ:
+            import errno
+            raise OSError(errno.EBADF, "peek() on write-only GzipFile object")
+
+        # Do not return ridiculously small buffers, for one common idiom
+        # is to call peek(1) and expect more bytes in return.
+        if n < 100:
+            n = 100
+        if self.extrasize == 0:
+            if self.fileobj is None:
+                return b''
+            # Ensure that we don't return b"" if we haven't reached EOF.
+            # 1024 is the same buffering heuristic used in read()
+            while self.extrasize == 0 and self._read(max(n, 1024)):
+                pass
+        offset = self.offset - self.extrastart
+        remaining = self.extrasize
+        assert remaining == len(self.extrabuf) - offset
+        return self.extrabuf[offset:offset + n]
+
+    def _unread(self, buf):
+        self.extrasize = len(buf) + self.extrasize
+        self.offset -= len(buf)
+
+    def _read(self, size=1024):
+        if self.fileobj is None:
+            return False
+
+        if self._new_member:
+            # If the _new_member flag is set, we have to
+            # jump to the next member, if there is one.
+            self._init_read()
+            if not self._read_gzip_header():
+                return False
+            self.decompress = zlib.decompressobj(-zlib.MAX_WBITS)
+            self._new_member = False
+
+        # Read a chunk of data from the file
+        buf = self.fileobj.read(size)
+
+        # If the EOF has been reached, flush the decompression object
+        # and mark this object as finished.
+
+        if buf == b"":
+            uncompress = self.decompress.flush()
+            # Prepend the already read bytes to the fileobj to they can be
+            # seen by _read_eof()
+            self.fileobj.prepend(self.decompress.unused_data, True)
+            self._read_eof()
+            self._add_read_data(uncompress)
+            return False
+
+        uncompress = self.decompress.decompress(buf)
+        self._add_read_data(uncompress)
+
+        if self.decompress.unused_data != b"":
+            # Ending case: we've come to the end of a member in the file,
+            # so seek back to the start of the unused data, finish up
+            # this member, and read a new gzip header.
+            # Prepend the already read bytes to the fileobj to they can be
+            # seen by _read_eof() and _read_gzip_header()
+            self.fileobj.prepend(self.decompress.unused_data, True)
+            # Check the CRC and file size, and set the flag so we read
+            # a new member on the next call
+            self._read_eof()
+            self._new_member = True
+        return True
+
+    def _add_read_data(self, data):
+        self.crc = zlib.crc32(data, self.crc) & 0xffffffff
+        offset = self.offset - self.extrastart
+        self.extrabuf = self.extrabuf[offset:] + data
+        self.extrasize = self.extrasize + len(data)
+        self.extrastart = self.offset
+        self.size = self.size + len(data)
+
+    def _read_eof(self):
+        # We've read to the end of the file
+        # We check the that the computed CRC and size of the
+        # uncompressed data matches the stored values.  Note that the size
+        # stored is the true file size mod 2**32.
+        crc32, isize = struct.unpack("<II", self._read_exact(8))
+        if crc32 != self.crc:
+            raise OSError("CRC check failed %s != %s" % (hex(crc32),
+                                                         hex(self.crc)))
+        elif isize != (self.size & 0xffffffff):
+            raise OSError("Incorrect length of data produced")
+
+        # Gzip files can be padded with zeroes and still have archives.
+        # Consume all zero bytes and set the file position to the first
+        # non-zero byte. See http://www.gzip.org/#faq8
+        c = b"\x00"
+        while c == b"\x00":
+            c = self.fileobj.read(1)
+        if c:
+            self.fileobj.prepend(c, True)
+
+    @property
+    def closed(self):
+        return self.fileobj is None
+
+    def close(self):
+        fileobj = self.fileobj
+        if fileobj is None:
+            return
+        self.fileobj = None
+        try:
+            if self.mode == WRITE:
+                fileobj.write(self.compress.flush())
+                write32u(fileobj, self.crc)
+                # self.size may exceed 2GB, or even 4GB
+                write32u(fileobj, self.size & 0xffffffff)
+        finally:
+            myfileobj = self.myfileobj
+            if myfileobj:
+                self.myfileobj = None
+                myfileobj.close()
+
+    def flush(self, zlib_mode=zlib.Z_SYNC_FLUSH):
+        self._check_closed()
+        if self.mode == WRITE:
+            # Ensure the compressor's buffer is flushed
+            self.fileobj.write(self.compress.flush(zlib_mode))
+            self.fileobj.flush()
+
+    def fileno(self):
+        """Invoke the underlying file object's fileno() method.
+
+        This will raise AttributeError if the underlying file object
+        doesn't support fileno().
+        """
+        return self.fileobj.fileno()
+
+    def rewind(self):
+        '''Return the uncompressed stream file position indicator to the
+        beginning of the file'''
+        if self.mode != READ:
+            raise OSError("Can't rewind in write mode")
+        self.fileobj.seek(0)
+        self._new_member = True
+        self.extrabuf = b""
+        self.extrasize = 0
+        self.extrastart = 0
+        self.offset = 0
+
+    def readable(self):
+        return self.mode == READ
+
+    def writable(self):
+        return self.mode == WRITE
+
+    def seekable(self):
+        return True
+
+    def seek(self, offset, whence=0):
+        if whence:
+            if whence == 1:
+                offset = self.offset + offset
+            else:
+                raise ValueError('Seek from end not supported')
+        if self.mode == WRITE:
+            if offset < self.offset:
+                raise OSError('Negative seek in write mode')
+            count = offset - self.offset
+            chunk = bytes(1024)
+            for i in range(count // 1024):
+                self.write(chunk)
+            self.write(bytes(count % 1024))
+        elif self.mode == READ:
+            if offset < self.offset:
+                # for negative seek, rewind and do positive seek
+                self.rewind()
+            count = offset - self.offset
+            for i in range(count // 1024):
+                self.read(1024)
+            self.read(count % 1024)
+
+        return self.offset
+
+    def readline(self, size=-1):
+        if size < 0:
+            # Shortcut common case - newline found in buffer.
+            offset = self.offset - self.extrastart
+            i = self.extrabuf.find(b'\n', offset) + 1
+            if i > 0:
+                self.extrasize -= i - offset
+                self.offset += i - offset
+                return self.extrabuf[offset: i]
+
+            size = sys.maxsize
+            readsize = self.min_readsize
+        else:
+            readsize = size
+        bufs = []
+        while size != 0:
+            c = self.read(readsize)
+            i = c.find(b'\n')
+
+            # We set i=size to break out of the loop under two
+            # conditions: 1) there's no newline, and the chunk is
+            # larger than size, or 2) there is a newline, but the
+            # resulting line would be longer than 'size'.
+            if (size <= i) or (i == -1 and len(c) > size):
+                i = size - 1
+
+            if i >= 0 or c == b'':
+                bufs.append(c[:i + 1])    # Add portion of last chunk
+                self._unread(c[i + 1:])   # Push back rest of chunk
+                break
+
+            # Append chunk to list, decrease 'size',
+            bufs.append(c)
+            size = size - len(c)
+            readsize = min(size, readsize * 2)
+        if readsize > self.min_readsize:
+            self.min_readsize = min(readsize, self.min_readsize * 2, 512)
+        return b''.join(bufs)  # Return resulting line
+
+
+def compress(data, compresslevel=9):
+    """Compress data in one shot and return the compressed string.
+    Optional argument is the compression level, in range of 0-9.
+    """
+    buf = io.BytesIO()
+    with GzipFile(fileobj=buf, mode='wb', compresslevel=compresslevel) as f:
+        f.write(data)
+    return buf.getvalue()
+
+
+def decompress(data):
+    """Decompress a gzip compressed string in one shot.
+    Return the decompressed string.
+    """
+    with GzipFile(fileobj=io.BytesIO(data)) as f:
+        return f.read()
diff --git a/apitools/base/py/gzip_test.py b/apitools/base/py/gzip_test.py
new file mode 100644
index 00000000..2d7d4580
--- /dev/null
+++ b/apitools/base/py/gzip_test.py
@@ -0,0 +1,514 @@
+# Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+# 2011, 2012, 2013, 2014, 2015, 2016, 2017 Python Software Foundation; All
+# Rights Reserved
+#
+# This is a backport from python 3.4 into python 2.7. Text and exclusive mode
+# support are removed as they're unsupported in 2.7. This backport patches a
+# streaming bug that exists in python 2.7.
+
+"""Test script for the gzip module.
+"""
+
+import six
+from six.moves import range
+
+import unittest
+import os
+import io
+import struct
+from apitools.base.py import gzip
+from io import open
+
+data1 = b"""  int length=DEFAULTALLOC, err = Z_OK;
+  PyObject *RetVal;
+  int flushmode = Z_FINISH;
+  unsigned long start_total_out;
+
+"""
+
+data2 = b"""/* zlibmodule.c -- gzip-compatible data compression */
+/* See http://www.gzip.org/zlib/
+/* See http://www.winimage.com/zLibDll for Windows */
+"""
+
+
+def unlink(filename):
+    try:
+        os.unlink(filename)
+    except:
+        pass
+
+
+class UnseekableIO(io.BytesIO):
+    def seekable(self):
+        return False
+
+    def tell(self):
+        raise io.UnsupportedOperation
+
+    def seek(self, *args):
+        raise io.UnsupportedOperation
+
+
+class BaseTest(unittest.TestCase):
+    filename = "@test"
+
+    def setUp(self):
+        unlink(self.filename)
+
+    def tearDown(self):
+        unlink(self.filename)
+
+
+class TestGzip(BaseTest):
+    def write_and_read_back(self, data, mode='b'):
+        b_data = bytes(data)
+        with gzip.GzipFile(self.filename, 'w' + mode) as f:
+            l = f.write(data)
+        self.assertEqual(l, len(b_data))
+        with gzip.GzipFile(self.filename, 'r' + mode) as f:
+            self.assertEqual(f.read(), b_data)
+
+    def test_write(self):
+        with gzip.GzipFile(self.filename, 'wb') as f:
+            f.write(data1 * 50)
+
+            # Try flush and fileno.
+            f.flush()
+            f.fileno()
+            if hasattr(os, 'fsync'):
+                os.fsync(f.fileno())
+            f.close()
+
+        # Test multiple close() calls.
+        f.close()
+
+    # The following test_write_xy methods test that write accepts
+    # the corresponding bytes-like object type as input
+    # and that the data written equals bytes(xy) in all cases.
+    def test_write_memoryview(self):
+        data = memoryview(data1 * 50)
+        self.write_and_read_back(data.tobytes())
+        data = memoryview(bytes(range(256)))
+        self.write_and_read_back(data.tobytes())
+
+    def test_write_incompatible_type(self):
+        # Test that non-bytes-like types raise TypeError.
+        # Issue #21560: attempts to write incompatible types
+        # should not affect the state of the fileobject
+        with gzip.GzipFile(self.filename, 'wb') as f:
+            if six.PY2:
+                with self.assertRaises(UnicodeEncodeError):
+                    f.write(u'\xff')
+            elif six.PY3:
+                with self.assertRaises(TypeError):
+                    f.write(u'\xff')
+            with self.assertRaises(TypeError):
+                f.write([1])
+            f.write(data1)
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            self.assertEqual(f.read(), data1)
+
+    def test_read(self):
+        self.test_write()
+        # Try reading.
+        with gzip.GzipFile(self.filename, 'r') as f:
+            d = f.read()
+        self.assertEqual(d, data1 * 50)
+
+    def test_read1(self):
+        self.test_write()
+        blocks = []
+        nread = 0
+        with gzip.GzipFile(self.filename, 'r') as f:
+            while True:
+                d = f.read1()
+                if not d:
+                    break
+                blocks.append(d)
+                nread += len(d)
+                # Check that position was updated correctly (see issue10791).
+                self.assertEqual(f.tell(), nread)
+        self.assertEqual(b''.join(blocks), data1 * 50)
+
+    def test_io_on_closed_object(self):
+        # Test that I/O operations on closed GzipFile objects raise a
+        # ValueError, just like the corresponding functions on file objects.
+
+        # Write to a file, open it for reading, then close it.
+        self.test_write()
+        f = gzip.GzipFile(self.filename, 'r')
+        f.close()
+        with self.assertRaises(ValueError):
+            f.read(1)
+        with self.assertRaises(ValueError):
+            f.seek(0)
+        with self.assertRaises(ValueError):
+            f.tell()
+        # Open the file for writing, then close it.
+        f = gzip.GzipFile(self.filename, 'w')
+        f.close()
+        with self.assertRaises(ValueError):
+            f.write(b'')
+        with self.assertRaises(ValueError):
+            f.flush()
+
+    def test_append(self):
+        self.test_write()
+        # Append to the previous file
+        with gzip.GzipFile(self.filename, 'ab') as f:
+            f.write(data2 * 15)
+
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            d = f.read()
+        self.assertEqual(d, (data1 * 50) + (data2 * 15))
+
+    def test_many_append(self):
+        # Bug #1074261 was triggered when reading a file that contained
+        # many, many members.  Create such a file and verify that reading it
+        # works.
+        with gzip.GzipFile(self.filename, 'wb', 9) as f:
+            f.write(b'a')
+        for i in range(0, 200):
+            with gzip.GzipFile(self.filename, "ab", 9) as f:  # append
+                f.write(b'a')
+
+        # Try reading the file
+        with gzip.GzipFile(self.filename, "rb") as zgfile:
+            contents = b""
+            while 1:
+                ztxt = zgfile.read(8192)
+                contents += ztxt
+                if not ztxt:
+                    break
+        self.assertEqual(contents, b'a' * 201)
+
+    def test_buffered_reader(self):
+        # Issue #7471: a GzipFile can be wrapped in a BufferedReader for
+        # performance.
+        self.test_write()
+
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            with io.BufferedReader(f) as r:
+                lines = [line for line in r]
+
+        self.assertEqual(lines, 50 * data1.splitlines(True))
+
+    def test_readline(self):
+        self.test_write()
+        # Try .readline() with varying line lengths
+
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            line_length = 0
+            while 1:
+                L = f.readline(line_length)
+                if not L and line_length != 0:
+                    break
+                self.assertTrue(len(L) <= line_length)
+                line_length = (line_length + 1) % 50
+
+    def test_readlines(self):
+        self.test_write()
+        # Try .readlines()
+
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            L = f.readlines()
+
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            while 1:
+                L = f.readlines(150)
+                if L == []:
+                    break
+
+    def test_seek_read(self):
+        self.test_write()
+        # Try seek, read test
+
+        with gzip.GzipFile(self.filename) as f:
+            while 1:
+                oldpos = f.tell()
+                line1 = f.readline()
+                if not line1:
+                    break
+                newpos = f.tell()
+                f.seek(oldpos)  # negative seek
+                if len(line1) > 10:
+                    amount = 10
+                else:
+                    amount = len(line1)
+                line2 = f.read(amount)
+                self.assertEqual(line1[:amount], line2)
+                f.seek(newpos)  # positive seek
+
+    def test_seek_whence(self):
+        self.test_write()
+        # Try seek(whence=1), read test
+
+        with gzip.GzipFile(self.filename) as f:
+            f.read(10)
+            f.seek(10, whence=1)
+            y = f.read(10)
+        self.assertEqual(y, data1[20:30])
+
+    def test_seek_write(self):
+        # Try seek, write test
+        with gzip.GzipFile(self.filename, 'w') as f:
+            for pos in range(0, 256, 16):
+                f.seek(pos)
+                f.write(b'GZ\n')
+
+    def test_mode(self):
+        self.test_write()
+        with gzip.GzipFile(self.filename, 'r') as f:
+            self.assertEqual(f.myfileobj.mode, 'rb')
+
+    def test_1647484(self):
+        for mode in ('wb', 'rb'):
+            with gzip.GzipFile(self.filename, mode) as f:
+                self.assertTrue(hasattr(f, "name"))
+                self.assertEqual(f.name, self.filename)
+
+    def test_paddedfile_getattr(self):
+        self.test_write()
+        with gzip.GzipFile(self.filename, 'rb') as f:
+            self.assertTrue(hasattr(f.fileobj, "name"))
+            self.assertEqual(f.fileobj.name, self.filename)
+
+    def test_mtime(self):
+        mtime = 123456789
+        with gzip.GzipFile(self.filename, 'w', mtime=mtime) as fWrite:
+            fWrite.write(data1)
+        with gzip.GzipFile(self.filename) as fRead:
+            dataRead = fRead.read()
+            self.assertEqual(dataRead, data1)
+            self.assertTrue(hasattr(fRead, 'mtime'))
+            self.assertEqual(fRead.mtime, mtime)
+
+    def test_metadata(self):
+        mtime = 123456789
+
+        with gzip.GzipFile(self.filename, 'w', mtime=mtime) as fWrite:
+            fWrite.write(data1)
+
+        with open(self.filename, 'rb') as fRead:
+            # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
+
+            idBytes = fRead.read(2)
+            self.assertEqual(idBytes, b'\x1f\x8b')  # gzip ID
+
+            cmByte = fRead.read(1)
+            self.assertEqual(cmByte, b'\x08')  # deflate
+
+            flagsByte = fRead.read(1)
+            self.assertEqual(flagsByte, b'\x08')  # only the FNAME flag is set
+
+            mtimeBytes = fRead.read(4)
+            self.assertEqual(mtimeBytes, struct.pack(
+                '<i', mtime))  # little-endian
+
+            xflByte = fRead.read(1)
+            self.assertEqual(xflByte, b'\x02')  # maximum compression
+
+            osByte = fRead.read(1)
+            self.assertEqual(osByte, b'\xff')  # OS "unknown" (OS-independent)
+
+            # Since the FNAME flag is set, the zero-terminated filename
+            # follows. RFC 1952 specifies that this is the name of the input
+            # file, if any. However, the gzip module defaults to storing the
+            # name of the output file in this field.
+            expected = self.filename.encode('Latin-1') + b'\x00'
+            nameBytes = fRead.read(len(expected))
+            self.assertEqual(nameBytes, expected)
+
+            # Since no other flags were set, the header ends here. Rather than
+            # process the compressed data, let's seek to the trailer.
+            fRead.seek(os.stat(self.filename).st_size - 8)
+
+            crc32Bytes = fRead.read(4)  # CRC32 of uncompressed data [data1]
+            self.assertEqual(crc32Bytes, b'\xaf\xd7d\x83')
+
+            isizeBytes = fRead.read(4)
+            self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))
+
+    def test_with_open(self):
+        # GzipFile supports the context management protocol
+        with gzip.GzipFile(self.filename, "wb") as f:
+            f.write(b"xxx")
+        f = gzip.GzipFile(self.filename, "rb")
+        f.close()
+        try:
+            with f:
+                pass
+        except ValueError:
+            pass
+        else:
+            self.fail("__enter__ on a closed file didn't raise an exception")
+        try:
+            with gzip.GzipFile(self.filename, "wb") as f:
+                1 / 0
+        except ZeroDivisionError:
+            pass
+        else:
+            self.fail("1/0 didn't raise an exception")
+
+    def test_zero_padded_file(self):
+        with gzip.GzipFile(self.filename, "wb") as f:
+            f.write(data1 * 50)
+
+        # Pad the file with zeroes
+        with open(self.filename, "ab") as f:
+            f.write(b"\x00" * 50)
+
+        with gzip.GzipFile(self.filename, "rb") as f:
+            d = f.read()
+            self.assertEqual(d, data1 * 50, "Incorrect data in file")
+
+    def test_non_seekable_file(self):
+        uncompressed = data1 * 50
+        buf = UnseekableIO()
+        with gzip.GzipFile(fileobj=buf, mode="wb") as f:
+            f.write(uncompressed)
+        compressed = buf.getvalue()
+        buf = UnseekableIO(compressed)
+        with gzip.GzipFile(fileobj=buf, mode="rb") as f:
+            self.assertEqual(f.read(), uncompressed)
+
+    def test_peek(self):
+        uncompressed = data1 * 200
+        with gzip.GzipFile(self.filename, "wb") as f:
+            f.write(uncompressed)
+
+        def sizes():
+            while True:
+                for n in range(5, 50, 10):
+                    yield n
+
+        with gzip.GzipFile(self.filename, "rb") as f:
+            f.max_read_chunk = 33
+            nread = 0
+            for n in sizes():
+                s = f.peek(n)
+                if s == b'':
+                    break
+                self.assertEqual(f.read(len(s)), s)
+                nread += len(s)
+            self.assertEqual(f.read(100), b'')
+            self.assertEqual(nread, len(uncompressed))
+
+    def test_textio_readlines(self):
+        # Issue #10791: TextIOWrapper.readlines() fails when wrapping GzipFile.
+        lines = (data1 * 50).decode("ascii").splitlines(True)
+        self.test_write()
+        with gzip.GzipFile(self.filename, 'r') as f:
+            with io.TextIOWrapper(f, encoding="ascii") as t:
+                self.assertEqual(t.readlines(), lines)
+
+    def test_fileobj_from_fdopen(self):
+        # Issue #13781: Opening a GzipFile for writing fails when using a
+        # fileobj created with os.fdopen().
+        fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT)
+        with os.fdopen(fd, "wb") as f:
+            with gzip.GzipFile(fileobj=f, mode="w") as g:
+                pass
+
+    def test_bytes_filename(self):
+        str_filename = self.filename
+        try:
+            bytes_filename = str_filename.encode("ascii")
+        except UnicodeEncodeError:
+            self.skipTest("Temporary file name needs to be ASCII")
+        with gzip.GzipFile(bytes_filename, "wb") as f:
+            f.write(data1 * 50)
+        with gzip.GzipFile(bytes_filename, "rb") as f:
+            self.assertEqual(f.read(), data1 * 50)
+        # Sanity check that we are actually operating on the right file.
+        with gzip.GzipFile(str_filename, "rb") as f:
+            self.assertEqual(f.read(), data1 * 50)
+
+    # Testing compress/decompress shortcut functions
+
+    def test_compress(self):
+        for data in [data1, data2]:
+            for args in [(), (1,), (6,), (9,)]:
+                datac = gzip.compress(data, *args)
+                self.assertEqual(type(datac), bytes)
+                with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f:
+                    self.assertEqual(f.read(), data)
+
+    def test_decompress(self):
+        for data in (data1, data2):
+            buf = io.BytesIO()
+            with gzip.GzipFile(fileobj=buf, mode="wb") as f:
+                f.write(data)
+            self.assertEqual(gzip.decompress(buf.getvalue()), data)
+            # Roundtrip with compress
+            datac = gzip.compress(data)
+            self.assertEqual(gzip.decompress(datac), data)
+
+    def test_read_truncated(self):
+        data = data1 * 50
+        # Drop the CRC (4 bytes) and file size (4 bytes).
+        truncated = gzip.compress(data)[:-8]
+        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
+            self.assertRaises(EOFError, f.read)
+        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
+            self.assertEqual(f.read(len(data)), data)
+            self.assertRaises(EOFError, f.read, 1)
+        # Incomplete 10-byte header.
+        for i in range(2, 10):
+            with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f:
+                self.assertRaises(EOFError, f.read, 1)
+
+    def test_read_with_extra(self):
+        # Gzip data with an extra field
+        gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff'
+                  b'\x05\x00Extra'
+                  b'\x0bI-.\x01\x002\xd1Mx\x04\x00\x00\x00')
+        with gzip.GzipFile(fileobj=io.BytesIO(gzdata)) as f:
+            self.assertEqual(f.read(), b'Test')
+
+    def test_prepend_error(self):
+        # See issue #20875
+        with gzip.open(self.filename, "wb") as f:
+            f.write(data1)
+        with gzip.open(self.filename, "rb") as f:
+            f.fileobj.prepend()
+
+
+class TestOpen(BaseTest):
+    def test_binary_modes(self):
+        uncompressed = data1 * 50
+
+        with gzip.open(self.filename, "wb") as f:
+            f.write(uncompressed)
+        with open(self.filename, "rb") as f:
+            file_data = gzip.decompress(f.read())
+            self.assertEqual(file_data, uncompressed)
+
+        with gzip.open(self.filename, "rb") as f:
+            self.assertEqual(f.read(), uncompressed)
+
+        with gzip.open(self.filename, "ab") as f:
+            f.write(uncompressed)
+        with open(self.filename, "rb") as f:
+            file_data = gzip.decompress(f.read())
+            self.assertEqual(file_data, uncompressed * 2)
+
+    def test_implicit_binary_modes(self):
+        # Test implicit binary modes (no "b" or "t" in mode string).
+        uncompressed = data1 * 50
+
+        with gzip.open(self.filename, "w") as f:
+            f.write(uncompressed)
+        with open(self.filename, "rb") as f:
+            file_data = gzip.decompress(f.read())
+            self.assertEqual(file_data, uncompressed)
+
+        with gzip.open(self.filename, "r") as f:
+            self.assertEqual(f.read(), uncompressed)
+
+        with gzip.open(self.filename, "a") as f:
+            f.write(uncompressed)
+        with open(self.filename, "rb") as f:
+            file_data = gzip.decompress(f.read())
+            self.assertEqual(file_data, uncompressed * 2)
diff --git a/apitools/base/py/transfer.py b/apitools/base/py/transfer.py
index 3a5ca93c..e6b5df63 100644
--- a/apitools/base/py/transfer.py
+++ b/apitools/base/py/transfer.py
@@ -30,6 +30,7 @@
 from six.moves import http_client
 
 from apitools.base.py import buffered_stream
+from apitools.base.py import compression
 from apitools.base.py import exceptions
 from apitools.base.py import http_wrapper
 from apitools.base.py import stream_slice
@@ -865,7 +866,8 @@ def __ValidateChunksize(self, chunksize=None):
                 self.__server_chunk_granularity)
 
     def __StreamMedia(self, callback=None, finish_callback=None,
-                      additional_headers=None, use_chunks=True):
+                      additional_headers=None, use_chunks=True,
+                      compressed=False):
         """Helper function for StreamMedia / StreamInChunks."""
         if self.strategy != RESUMABLE_UPLOAD:
             raise exceptions.InvalidUserInputError(
@@ -874,13 +876,22 @@ def __StreamMedia(self, callback=None, finish_callback=None,
         finish_callback = finish_callback or self.finish_callback
         # final_response is set if we resumed an already-completed upload.
         response = self.__final_response
-        send_func = self.__SendChunk if use_chunks else self.__SendMediaBody
+
+        def CallSendChunk(start):
+            return self.__SendChunk(
+                start, additional_headers=additional_headers,
+                compressed=compressed)
+
+        def CallSendMediaBody(start):
+            return self.__SendMediaBody(
+                start, additional_headers=additional_headers)
+
+        send_func = CallSendChunk if use_chunks else CallSendMediaBody
         if use_chunks:
             self.__ValidateChunksize(self.chunksize)
         self.EnsureInitialized()
         while not self.complete:
-            response = send_func(self.stream.tell(),
-                                 additional_headers=additional_headers)
+            response = send_func(self.stream.tell())
             if response.status_code in (http_client.OK, http_client.CREATED):
                 self.__complete = True
                 break
@@ -923,11 +934,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):
+                       additional_headers=None, compressed=False):
         """Send this (resumable) upload in chunks."""
         return self.__StreamMedia(
             callback=callback, finish_callback=finish_callback,
-            additional_headers=additional_headers)
+            additional_headers=additional_headers, compressed=compressed)
 
     def __SendMediaRequest(self, request, end):
         """Request helper function for SendMediaBody & SendChunk."""
@@ -972,11 +983,21 @@ def __SendMediaBody(self, start, additional_headers=None):
 
         return self.__SendMediaRequest(request, self.total_size)
 
-    def __SendChunk(self, start, additional_headers=None):
+    def __SendChunk(self, start, additional_headers=None, compressed=False):
         """Send the specified chunk."""
         self.EnsureInitialized()
         no_log_body = self.total_size is None
-        if self.total_size is None:
+        request = http_wrapper.Request(url=self.url, http_method='PUT')
+        if compressed:
+            request.headers['Content-Encoding'] = 'gzip'
+            body_stream, read_length, exhausted = compression.CompressStream(
+                self.stream, self.chunksize)
+            end = start + read_length
+            # If the stream length was previously unknown and the input stream
+            # is exhausted, then we're at the end of the stream.
+            if self.total_size is None and exhausted:
+                self.__total_size = end
+        elif self.total_size is None:
             # For the streaming resumable case, we need to detect when
             # we're at the end of the stream.
             body_stream = buffered_stream.BufferedStream(
@@ -995,8 +1016,7 @@ def __SendChunk(self, start, additional_headers=None):
             body_stream = stream_slice.StreamSlice(self.stream, end - start)
         # TODO(craigcitro): Think about clearer errors on "no data in
         # stream".
-        request = http_wrapper.Request(url=self.url, http_method='PUT',
-                                       body=body_stream)
+        request.body = body_stream
         request.headers['Content-Type'] = self.mime_type
         if no_log_body:
             # Disable logging of streaming body.
diff --git a/apitools/base/py/transfer_test.py b/apitools/base/py/transfer_test.py
index e738e08d..2b22caa5 100644
--- a/apitools/base/py/transfer_test.py
+++ b/apitools/base/py/transfer_test.py
@@ -150,7 +150,7 @@ def testGetRange(self):
                                          (start_byte, end_byte),
                         'status': http_client.OK,
                     },
-                    content=string.ascii_lowercase[start_byte:end_byte+1],
+                    content=string.ascii_lowercase[start_byte:end_byte + 1],
                     request_url=base_url,
                 )
                 request = http_wrapper.Request(url='https://part.one/')
@@ -306,3 +306,42 @@ def testMultipartEncoding(self):
             self.assertEqual(url_builder.query_params['uploadType'], 'media')
             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."""
+
+        # Sample highly compressible data.
+        sample_data = b'abc' * 100
+        sample_response = http_wrapper.Response(
+            info={'status': http_client.OK},
+            content='',
+            request_url='url',
+        )
+
+        # 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)
+
+            # 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)
+            # 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))
diff --git a/run_pylint.py b/run_pylint.py
index c53943fa..c7e532ec 100644
--- a/run_pylint.py
+++ b/run_pylint.py
@@ -45,6 +45,8 @@
     'ez_setup.py',
     'run_pylint.py',
     'setup.py',
+    'apitools/base/py/gzip.py',
+    'apitools/base/py/gzip_test.py',
 ]
 PRODUCTION_RC = 'default.pylintrc'
 TEST_RC = 'reduced.pylintrc'