path: root/paste/wsgilib.py

# (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

"""
A module of many disparate routines.
"""

# functions which moved to paste.request and paste.response
# Deprecated around 15 Dec 2005
from request import get_cookies, parse_querystring, parse_formvars
from request import construct_url, path_info_split, path_info_pop
from response import HeaderDict, has_header, header_value, remove_header
from response import error_body_response, error_response, error_response_app

from Cookie import SimpleCookie
from cStringIO import StringIO
import mimetypes
import os
import cgi
import sys
import re
from urlparse import urlsplit
import warnings

__all__ = ['get_cookies', 'add_close', 'raw_interactive',
           'interactive', 'construct_url', 'error_body_response',
           'error_response', 'send_file', 'has_header', 'header_value',
           'path_info_split', 'path_info_pop', 'capture_output',
           'catch_errors', 'dump_environ']


class add_close:
    """
    An an iterable that iterates over app_iter, then calls
    close_func.
    """

    def __init__(self, app_iterable, close_func):
        self.app_iterable = app_iterable
        self.app_iter = iter(app_iterable)
        self.close_func = close_func

    def __iter__(self):
        return self

    def next(self):
        return self.app_iter.next()

    def close(self):
        if hasattr(self.app_iterable, 'close'):
            self.app_iterable.close()
        self.close_func()

def catch_errors(application, environ, start_response, error_callback,
                 ok_callback=None):
    """
    Runs the application, and returns the application iterator (which should be
    passed upstream).  If an error occurs then error_callback will be called with
    exc_info as its sole argument.  If no errors occur and ok_callback is given,
    then it will be called with no arguments.
    """
    error_occurred = False
    try:
        app_iter = application(environ, start_response)
    except:
        error_callback(sys.exc_info())
        raise
    if type(app_iter) in (list, tuple):
        # These won't produce exceptions
        if ok_callback:
            ok_callback()
        return app_iter
    else:
        return _wrap_app_iter(app_iter, error_callback, ok_callback)

class _wrap_app_iter(object):

    def __init__(self, app_iterable, error_callback, ok_callback):
        self.app_iterable = app_iterable
        self.app_iter = iter(app_iterable)
        self.error_callback = error_callback
        self.ok_callback = ok_callback
        if hasattr(self.app_iterable, 'close'):
            self.close = self.app_iterable.close

    def __iter__(self):
        return self

    def next(self):
        try:
            return self.app_iter.next()
        except StopIteration:
            if self.ok_callback:
                self.ok_callback()
            raise
        except:
            self.error_callback(sys.exc_info())
            raise

def catch_errors_app(application, environ, start_response, error_callback_app,
                     ok_callback=None, catch=Exception):
    """
    Like ``catch_errors``, except error_callback_app should be a
    callable that will receive *three* arguments -- ``environ``,
    ``start_response``, and ``exc_info``.  It should call
    ``start_response`` (*with* the exc_info argument!) and return an
    iterator.
    """
    error_occurred = False
    try:
        app_iter = application(environ, start_response)
    except catch:
        return error_callback_app(environ, start_response, sys.exc_info())
    if type(app_iter) in (list, tuple):
        # These won't produce exceptions
        if ok_callback:
            ok_callback()
        return app_iter
    else:
        return _wrap_app_iter_app(
            environ, start_response, app_iter,
            error_callback_app, ok_callback)

class _wrap_app_iter_app(object):

    def __init__(self, environ, start_response, app_iterable,
                 error_callback_app, ok_callback):
        self.environ = environ
        self.start_response = start_response
        self.app_iterable = app_iterable
        self.app_iter = iter(app_iterable)
        self.error_callback_app = error_callback_app
        self.ok_callback = ok_callback
        if hasattr(self.app_iterable, 'close'):
            self.close = self.app_iterable.close

    def __iter__(self):
        return self

    def next(self):
        try:
            return self.app_iter.next()
        except StopIteration:
            if self.ok_callback:
                self.ok_callback()
            raise
        except:
            if hasattr(self.app_iterable, 'close'):
                try:
                    self.app_iterable.close()
                except:
                    # @@: Print to wsgi.errors?
                    pass
            new_app_iterable = self.error_callback_app(
                self.environ, self.start_response, sys.exc_info())
            app_iter = iter(new_app_iterable)
            if hasattr(new_app_iterable, 'close'):
                self.close = new_app_iterable.close
            self.next = app_iter.next
            return self.next()

def raw_interactive(application, path='', **environ):
    """
    Runs the application in a fake environment.
    """
    assert "path_info" not in environ, "argument list changed"
    errors = StringIO()
    basic_environ = {
        # mandatory CGI variables
        'REQUEST_METHOD': 'GET',     # always mandatory
        'SCRIPT_NAME': '',           # may be empty if app is at the root
        'PATH_INFO': '',             # may be empty if at root of app
        'SERVER_NAME': 'localhost',  # always mandatory
        'SERVER_PORT': '80',         # always mandatory 
        'SERVER_PROTOCOL': 'HTTP/1.0',
        # mandatory wsgi variables
        'wsgi.version': (1, 0),
        'wsgi.url_scheme': 'http',
        'wsgi.input': StringIO(''),
        'wsgi.errors': errors,
        'wsgi.multithread': False,
        'wsgi.multiprocess': False,
        'wsgi.run_once': False,
        }
    if path:
        (_,_,path_info,query,fragment) = urlsplit(str(path))
        basic_environ['PATH_INFO'] = path_info
        if query:
            basic_environ['QUERY_STRING'] = query
    for name, value in environ.items():
        name = name.replace('__', '.')
        basic_environ[name] = value
    if ('SERVER_NAME' in basic_environ
        and 'HTTP_HOST' not in basic_environ):
        basic_environ['HTTP_HOST'] = basic_environ['SERVER_NAME']
    istream = basic_environ['wsgi.input']
    if isinstance(istream, str):
        basic_environ['wsgi.input'] = StringIO(istream)
        basic_environ['CONTENT_LENGTH'] = len(istream)
    data = {}
    output = StringIO()
    headers_set = []
    headers_sent = []
    def start_response(status, headers, exc_info=None):
        if exc_info:
            try:
                if headers_sent:
                    # Re-raise original exception only if headers sent
                    raise exc_info[0], exc_info[1], exc_info[2]
                else:
                    # We assume that the sender, who is probably setting
                    # the headers a second time /w a 500 has produced
                    # a more appropriate response.
                    pass
            finally:
                # avoid dangling circular reference
                exc_info = None
        elif headers_set:
            # You cannot set the headers more than once, unless the
            # exc_info is provided.
            raise AssertionError("Headers already set and no exc_info!")
        headers_set.append(True)
        data['status'] = status
        data['headers'] = headers
        return output.write
    app_iter = application(basic_environ, start_response)
    try:
        try:
            for s in app_iter:
                headers_sent.append(True)
                if not headers_set:
                    raise AssertionError("Content sent w/o headers!")
                output.write(s)
        except TypeError, e:
            # Typically "iteration over non-sequence", so we want
            # to give better debugging information...
            e.args = ((e.args[0] + ' iterable: %r' % app_iter),) + e.args[1:]
            raise
    finally:
        if hasattr(app_iter, 'close'):
            app_iter.close()
    return (data['status'], data['headers'], output.getvalue(),
            errors.getvalue())

def interactive(*args, **kw):
    """
    Runs the application interatively, wrapping `raw_interactive` but
    returning the output in a formatted way.
    """
    status, headers, content, errors = raw_interactive(*args, **kw)
    full = StringIO()
    if errors:
        full.write('Errors:\n')
        full.write(errors.strip())
        full.write('\n----------end errors\n')
    full.write(status + '\n')
    for name, value in headers:
        full.write('%s: %s\n' % (name, value))
    full.write('\n')
    full.write(content)
    return full.getvalue()
interactive.proxy = 'raw_interactive'

def dump_environ(environ,start_response):
    """
    Application which simply dumps the current environment
    variables out as a plain text response.
    """
    output = []
    keys = environ.keys()
    keys.sort()
    for k in keys:
        v = str(environ[k]).replace("\n","\n    ")
        output.append("%s: %s\n" % (k,v))
    output.append("\n")
    content_length = environ.get("CONTENT_LENGTH",'')
    if content_length:
        output.append(environ['wsgi.input'].read(int(content_length)))
        output.append("\n")
    output = "".join(output)
    headers = [('Content-Type', 'text/plain'),
               ('Content-Length', len(output))]
    start_response("200 OK",headers)
    return [output]

def send_file(filename):
    warnings.warn(
        "wsgilib.send_file has been moved to paste.fileapp.FileApp",
        DeprecationWarning, 2)
    import fileapp
    return fileapp.FileApp(filename)

def capture_output(environ, start_response, application):
    """
    Runs application with environ and start_response, and captures
    status, headers, and body.

    Sends status and header, but *not* body.  Returns (status,
    headers, body).  Typically this is used like::

        def dehtmlifying_middleware(application):
            def replacement_app(environ, start_response):
                status, headers, body = capture_output(
                    environ, start_response, application)
                content_type = header_value(headers, 'content-type')
                if (not content_type
                    or not content_type.startswith('text/html')):
                    return [body]
                body = re.sub(r'<.*?>', '', body)
                return [body]
            return replacement_app
    """
    warnings.warn(
        'wsgilib.capture_output has been deprecated in favor '
        'of wsgilib.intercept_output',
        DeprecationWarning, 2)
    data = []
    output = StringIO()
    def replacement_start_response(status, headers, exc_info=None):
        if data:
            data[:] = []
        data.append(status)
        data.append(headers)
        start_response(status, headers, exc_info)
        return output.write
    app_iter = application(environ, replacement_start_response)
    try:
        for item in app_iter:
            output.write(item)
    finally:
        if hasattr(app_iter, 'close'):
            app_iter.close()
    if not data:
        data.append(None)
    if len(data) < 2:
        data.append(None)
    data.append(output.getvalue())
    return data

def intercept_output(environ, application):
    """
    Runs application with environ and captures status, headers, and
    body.  None are sent on; you must send them on yourself (unlike
    ``capture_output``)

    Typically this is used like::

        def dehtmlifying_middleware(application):
            def replacement_app(environ, start_response):
                status, headers, body = capture_output(
                    environ, start_response, application)
                content_type = header_value(headers, 'content-type')
                if (not content_type
                    or not content_type.startswith('text/html')):
                    return [body]
                body = re.sub(r'<.*?>', '', body)
                return [body]
            return replacement_app
    """
    data = []
    output = StringIO()
    def replacement_start_response(status, headers, exc_info=None):
        if data:
            data[:] = []
        data.append(status)
        data.append(headers)
        return output.write
    app_iter = application(environ, replacement_start_response)
    try:
        for item in app_iter:
            output.write(item)
    finally:
        if hasattr(app_iter, 'close'):
            app_iter.close()
    if not data:
        data.append(None)
    if len(data) < 2:
        data.append(None)
    data.append(output.getvalue())
    return data

## Deprecation warning wrapper:

class ResponseHeaderDict(HeaderDict):

    def __init__(self, *args, **kw):
        warnings.warn(
            "The class wsgilib.ResponseHeaderDict has been moved "
            "to paste.response.ResponseHeaderDict",
            DeprecationWarning, 2)
        HeaderDict.__init__(self, *args, **kw)

def _warn_deprecated(new_func):
    new_name = new_func.func_name
    new_path = new_func.func_globals['__name__'] + '.' + new_name
    def replacement(*args, **kw):
        warnings.warn(
            "The function wsgilib.%s has been moved to %s"
            % (new_name, new_path),
            DeprecationWarning, 2)
        return new_func(*args, **kw)
    try:
        replacement.func_name = new_func.func_name
    except:
        pass
    return replacement

# Put warnings wrapper in place for all public functions that
# were imported from elsewhere:

for _name in __all__:
    _func = globals()[_name]
    if (hasattr(_func, 'func_globals')
        and _func.func_globals['__name__'] != __name__):
        globals()[_name] = _warn_deprecated(_func)

if __name__ == '__main__':
    import doctest
    doctest.testmod()