| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107 |
- # (c) 2005 Ian Bicking and contributors; written for Paste (http://pythonpaste.org)
- # Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php
- # (c) 2005 Ian Bicking and contributors; written for Paste (http://pythonpaste.org)
- # Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php
- """
- WSGI middleware
- Gzip-encodes the response.
- """
- import gzip
- from paste.response import header_value, remove_header
- from paste.httpheaders import CONTENT_LENGTH
- import six
- class GzipOutput(object):
- pass
- class middleware(object):
- def __init__(self, application, compress_level=6):
- self.application = application
- self.compress_level = int(compress_level)
- def __call__(self, environ, start_response):
- if 'gzip' not in environ.get('HTTP_ACCEPT_ENCODING', ''):
- # nothing for us to do, so this middleware will
- # be a no-op:
- return self.application(environ, start_response)
- response = GzipResponse(start_response, self.compress_level)
- app_iter = self.application(environ,
- response.gzip_start_response)
- if app_iter is not None:
- response.finish_response(app_iter)
- return response.write()
- class GzipResponse(object):
- def __init__(self, start_response, compress_level):
- self.start_response = start_response
- self.compress_level = compress_level
- self.buffer = six.BytesIO()
- self.compressible = False
- self.content_length = None
- def gzip_start_response(self, status, headers, exc_info=None):
- self.headers = headers
- ct = header_value(headers,'content-type')
- ce = header_value(headers,'content-encoding')
- self.compressible = False
- if ct and (ct.startswith('text/') or ct.startswith('application/')) \
- and 'zip' not in ct:
- self.compressible = True
- if ce:
- self.compressible = False
- if self.compressible:
- headers.append(('content-encoding', 'gzip'))
- remove_header(headers, 'content-length')
- self.headers = headers
- self.status = status
- return self.buffer.write
- def write(self):
- out = self.buffer
- out.seek(0)
- s = out.getvalue()
- out.close()
- return [s]
- def finish_response(self, app_iter):
- if self.compressible:
- output = gzip.GzipFile(mode='wb', compresslevel=self.compress_level,
- fileobj=self.buffer)
- else:
- output = self.buffer
- try:
- for s in app_iter:
- output.write(s)
- if self.compressible:
- output.close()
- finally:
- if hasattr(app_iter, 'close'):
- app_iter.close()
- content_length = self.buffer.tell()
- CONTENT_LENGTH.update(self.headers, content_length)
- self.start_response(self.status, self.headers)
- def filter_factory(application, **conf):
- import warnings
- warnings.warn(
- 'This function is deprecated; use make_gzip_middleware instead',
- DeprecationWarning, 2)
- def filter(application):
- return middleware(application)
- return filter
- def make_gzip_middleware(app, global_conf, compress_level=6):
- """
- Wrap the middleware, so that it applies gzipping to a response
- when it is supported by the browser and the content is of
- type ``text/*`` or ``application/*``
- """
- compress_level = int(compress_level)
- return middleware(app, compress_level=compress_level)
|