<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> from paste . request import get_cookies , parse_querystring , parse_formvars <EOL> from paste . request import construct_url , path_info_split , path_info_pop <EOL> from paste . response import HeaderDict , has_header , header_value , remove_header <EOL> from paste . response import error_body_response , error_response , error_response_app <EOL> from traceback import print_exception <EOL> import six <EOL> import sys <EOL> from six . moves import cStringIO as StringIO <EOL> from six . moves . urllib . parse import unquote , urlsplit <EOL> import warnings <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class add_close ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app_iterable , close_func ) : <EOL> self . app_iterable = app_iterable <EOL> self . app_iter = iter ( app_iterable ) <EOL> self . close_func = close_func <EOL> self . _closed = False <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> return self . app_iter . next ( ) <EOL> def close ( self ) : <EOL> self . _closed = True <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> self . app_iterable . close ( ) <EOL> self . close_func ( ) <EOL> def __del__ ( self ) : <EOL> if not self . _closed : <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % self . close_func , file = sys . stderr ) <EOL> class add_start_close ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app_iterable , start_func , close_func = None ) : <EOL> self . app_iterable = app_iterable <EOL> self . app_iter = iter ( app_iterable ) <EOL> self . first = True <EOL> self . start_func = start_func <EOL> self . close_func = close_func <EOL> self . _closed = False <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> if self . first : <EOL> self . start_func ( ) <EOL> self . first = False <EOL> return next ( self . app_iter ) <EOL> __next__ = next <EOL> def close ( self ) : <EOL> self . _closed = True <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> self . app_iterable . close ( ) <EOL> if self . close_func is not None : <EOL> self . close_func ( ) <EOL> def __del__ ( self ) : <EOL> if not self . _closed : <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % self . close_func , file = sys . stderr ) <EOL> class chained_app_iters ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * chained ) : <EOL> self . app_iters = chained <EOL> self . chained = [ iter ( item ) for item in chained ] <EOL> self . _closed = False <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> if len ( self . chained ) == <NUM_LIT:1> : <EOL> return self . chained [ <NUM_LIT:0> ] . next ( ) <EOL> else : <EOL> try : <EOL> return self . chained [ <NUM_LIT:0> ] . next ( ) <EOL> except StopIteration : <EOL> self . chained . pop ( <NUM_LIT:0> ) <EOL> return self . next ( ) <EOL> def close ( self ) : <EOL> self . _closed = True <EOL> got_exc = None <EOL> for app_iter in self . app_iters : <EOL> try : <EOL> if hasattr ( app_iter , '<STR_LIT>' ) : <EOL> app_iter . close ( ) <EOL> except : <EOL> got_exc = sys . exc_info ( ) <EOL> if got_exc : <EOL> six . reraise ( got_exc [ <NUM_LIT:0> ] , got_exc [ <NUM_LIT:1> ] , got_exc [ <NUM_LIT:2> ] ) <EOL> def __del__ ( self ) : <EOL> if not self . _closed : <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % self . close_func , file = sys . stderr ) <EOL> class encode_unicode_app_iter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app_iterable , encoding = sys . getdefaultencoding ( ) , <EOL> errors = '<STR_LIT:strict>' ) : <EOL> self . app_iterable = app_iterable <EOL> self . app_iter = iter ( app_iterable ) <EOL> self . encoding = encoding <EOL> self . errors = errors <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> content = next ( self . app_iter ) <EOL> if isinstance ( content , six . text_type ) : <EOL> content = content . encode ( self . encoding , self . errors ) <EOL> return content <EOL> __next__ = next <EOL> def close ( self ) : <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> self . app_iterable . close ( ) <EOL> def catch_errors ( application , environ , start_response , error_callback , <EOL> ok_callback = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> app_iter = application ( environ , start_response ) <EOL> except : <EOL> error_callback ( sys . exc_info ( ) ) <EOL> raise <EOL> if type ( app_iter ) in ( list , tuple ) : <EOL> if ok_callback : <EOL> ok_callback ( ) <EOL> return app_iter <EOL> else : <EOL> return _wrap_app_iter ( app_iter , error_callback , ok_callback ) <EOL> class _wrap_app_iter ( object ) : <EOL> def __init__ ( self , app_iterable , error_callback , ok_callback ) : <EOL> self . app_iterable = app_iterable <EOL> self . app_iter = iter ( app_iterable ) <EOL> self . error_callback = error_callback <EOL> self . ok_callback = ok_callback <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> self . close = self . app_iterable . close <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> try : <EOL> return self . app_iter . next ( ) <EOL> except StopIteration : <EOL> if self . ok_callback : <EOL> self . ok_callback ( ) <EOL> raise <EOL> except : <EOL> self . error_callback ( sys . exc_info ( ) ) <EOL> raise <EOL> def catch_errors_app ( application , environ , start_response , error_callback_app , <EOL> ok_callback = None , catch = Exception ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> app_iter = application ( environ , start_response ) <EOL> except catch : <EOL> return error_callback_app ( environ , start_response , sys . exc_info ( ) ) <EOL> if type ( app_iter ) in ( list , tuple ) : <EOL> if ok_callback is not None : <EOL> ok_callback ( ) <EOL> return app_iter <EOL> else : <EOL> return _wrap_app_iter_app ( <EOL> environ , start_response , app_iter , <EOL> error_callback_app , ok_callback , catch = catch ) <EOL> class _wrap_app_iter_app ( object ) : <EOL> def __init__ ( self , environ , start_response , app_iterable , <EOL> error_callback_app , ok_callback , catch = Exception ) : <EOL> self . environ = environ <EOL> self . start_response = start_response <EOL> self . app_iterable = app_iterable <EOL> self . app_iter = iter ( app_iterable ) <EOL> self . error_callback_app = error_callback_app <EOL> self . ok_callback = ok_callback <EOL> self . catch = catch <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> self . close = self . app_iterable . close <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> try : <EOL> return self . app_iter . next ( ) <EOL> except StopIteration : <EOL> if self . ok_callback : <EOL> self . ok_callback ( ) <EOL> raise <EOL> except self . catch : <EOL> if hasattr ( self . app_iterable , '<STR_LIT>' ) : <EOL> try : <EOL> self . app_iterable . close ( ) <EOL> except : <EOL> pass <EOL> new_app_iterable = self . error_callback_app ( <EOL> self . environ , self . start_response , sys . exc_info ( ) ) <EOL> app_iter = iter ( new_app_iterable ) <EOL> if hasattr ( new_app_iterable , '<STR_LIT>' ) : <EOL> self . close = new_app_iterable . close <EOL> self . next = app_iter . next <EOL> return self . next ( ) <EOL> def raw_interactive ( application , path = '<STR_LIT>' , raise_on_wsgi_error = False , <EOL> ** environ ) : <EOL> """<STR_LIT>""" <EOL> assert "<STR_LIT>" not in environ , "<STR_LIT>" <EOL> if raise_on_wsgi_error : <EOL> errors = ErrorRaiser ( ) <EOL> else : <EOL> errors = six . BytesIO ( ) <EOL> basic_environ = { <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> '<STR_LIT>' : '<STR_LIT:http>' , <EOL> '<STR_LIT>' : six . BytesIO ( ) , <EOL> '<STR_LIT>' : errors , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> } <EOL> if path : <EOL> ( _ , _ , path_info , query , fragment ) = urlsplit ( str ( path ) ) <EOL> path_info = unquote ( path_info ) <EOL> path_info , query = str ( path_info ) , str ( query ) <EOL> basic_environ [ '<STR_LIT>' ] = path_info <EOL> if query : <EOL> basic_environ [ '<STR_LIT>' ] = query <EOL> for name , value in environ . items ( ) : <EOL> name = name . replace ( '<STR_LIT>' , '<STR_LIT:.>' ) <EOL> basic_environ [ name ] = value <EOL> if ( '<STR_LIT>' in basic_environ <EOL> and '<STR_LIT>' not in basic_environ ) : <EOL> basic_environ [ '<STR_LIT>' ] = basic_environ [ '<STR_LIT>' ] <EOL> istream = basic_environ [ '<STR_LIT>' ] <EOL> if isinstance ( istream , bytes ) : <EOL> basic_environ [ '<STR_LIT>' ] = six . BytesIO ( istream ) <EOL> basic_environ [ '<STR_LIT>' ] = len ( istream ) <EOL> data = { } <EOL> output = [ ] <EOL> headers_set = [ ] <EOL> headers_sent = [ ] <EOL> def start_response ( status , headers , exc_info = None ) : <EOL> if exc_info : <EOL> try : <EOL> if headers_sent : <EOL> six . reraise ( exc_info [ <NUM_LIT:0> ] , exc_info [ <NUM_LIT:1> ] , exc_info [ <NUM_LIT:2> ] ) <EOL> finally : <EOL> exc_info = None <EOL> elif headers_set : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> headers_set . append ( True ) <EOL> data [ '<STR_LIT:status>' ] = status <EOL> data [ '<STR_LIT>' ] = headers <EOL> return output . append <EOL> app_iter = application ( basic_environ , start_response ) <EOL> try : <EOL> try : <EOL> for s in app_iter : <EOL> if not isinstance ( s , six . binary_type ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % s ) <EOL> headers_sent . append ( True ) <EOL> if not headers_set : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> output . append ( s ) <EOL> except TypeError as e : <EOL> e . args = ( ( e . args [ <NUM_LIT:0> ] + '<STR_LIT>' % app_iter ) , ) + e . args [ <NUM_LIT:1> : ] <EOL> raise <EOL> finally : <EOL> if hasattr ( app_iter , '<STR_LIT>' ) : <EOL> app_iter . close ( ) <EOL> return ( data [ '<STR_LIT:status>' ] , data [ '<STR_LIT>' ] , b'<STR_LIT>' . join ( output ) , <EOL> errors . getvalue ( ) ) <EOL> class ErrorRaiser ( object ) : <EOL> def flush ( self ) : <EOL> pass <EOL> def write ( self , value ) : <EOL> if not value : <EOL> return <EOL> raise AssertionError ( <EOL> "<STR_LIT>" % value ) <EOL> def writelines ( self , seq ) : <EOL> raise AssertionError ( <EOL> "<STR_LIT>" % list ( seq ) ) <EOL> def getvalue ( self ) : <EOL> return '<STR_LIT>' <EOL> def interactive ( * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> status , headers , content , errors = raw_interactive ( * args , ** kw ) <EOL> full = StringIO ( ) <EOL> if errors : <EOL> full . write ( '<STR_LIT>' ) <EOL> full . write ( errors . strip ( ) ) <EOL> full . write ( '<STR_LIT>' ) <EOL> full . write ( status + '<STR_LIT:\n>' ) <EOL> for name , value in headers : <EOL> full . write ( '<STR_LIT>' % ( name , value ) ) <EOL> full . write ( '<STR_LIT:\n>' ) <EOL> full . write ( content ) <EOL> return full . getvalue ( ) <EOL> interactive . proxy = '<STR_LIT>' <EOL> def dump_environ ( environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> output = [ ] <EOL> keys = list ( environ . keys ( ) ) <EOL> keys . sort ( ) <EOL> for k in keys : <EOL> v = str ( environ [ k ] ) . replace ( "<STR_LIT:\n>" , "<STR_LIT>" ) <EOL> output . append ( "<STR_LIT>" % ( k , v ) ) <EOL> output . append ( "<STR_LIT:\n>" ) <EOL> content_length = environ . get ( "<STR_LIT>" , '<STR_LIT>' ) <EOL> if content_length : <EOL> output . append ( environ [ '<STR_LIT>' ] . read ( int ( content_length ) ) ) <EOL> output . append ( "<STR_LIT:\n>" ) <EOL> output = "<STR_LIT>" . join ( output ) <EOL> if six . PY3 : <EOL> output = output . encode ( '<STR_LIT:utf8>' ) <EOL> headers = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , str ( len ( output ) ) ) ] <EOL> start_response ( "<STR_LIT>" , headers ) <EOL> return [ output ] <EOL> def send_file ( filename ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" , <EOL> DeprecationWarning , <NUM_LIT:2> ) <EOL> from paste import fileapp <EOL> return fileapp . FileApp ( filename ) <EOL> def capture_output ( environ , start_response , application ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> DeprecationWarning , <NUM_LIT:2> ) <EOL> data = [ ] <EOL> output = StringIO ( ) <EOL> def replacement_start_response ( status , headers , exc_info = None ) : <EOL> if data : <EOL> data [ : ] = [ ] <EOL> data . append ( status ) <EOL> data . append ( headers ) <EOL> start_response ( status , headers , exc_info ) <EOL> return output . write <EOL> app_iter = application ( environ , replacement_start_response ) <EOL> try : <EOL> for item in app_iter : <EOL> output . write ( item ) <EOL> finally : <EOL> if hasattr ( app_iter , '<STR_LIT>' ) : <EOL> app_iter . close ( ) <EOL> if not data : <EOL> data . append ( None ) <EOL> if len ( data ) < <NUM_LIT:2> : <EOL> data . append ( None ) <EOL> data . append ( output . getvalue ( ) ) <EOL> return data <EOL> def intercept_output ( environ , application , conditional = None , <EOL> start_response = None ) : <EOL> """<STR_LIT>""" <EOL> if conditional is not None and start_response is None : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> data = [ ] <EOL> output = StringIO ( ) <EOL> def replacement_start_response ( status , headers , exc_info = None ) : <EOL> if conditional is not None and not conditional ( status , headers ) : <EOL> data . append ( None ) <EOL> return start_response ( status , headers , exc_info ) <EOL> if data : <EOL> data [ : ] = [ ] <EOL> data . append ( status ) <EOL> data . append ( headers ) <EOL> return output . write <EOL> app_iter = application ( environ , replacement_start_response ) <EOL> if data [ <NUM_LIT:0> ] is None : <EOL> return ( None , None , app_iter ) <EOL> try : <EOL> for item in app_iter : <EOL> output . write ( item ) <EOL> finally : <EOL> if hasattr ( app_iter , '<STR_LIT>' ) : <EOL> app_iter . close ( ) <EOL> if not data : <EOL> data . append ( None ) <EOL> if len ( data ) < <NUM_LIT:2> : <EOL> data . append ( None ) <EOL> data . append ( output . getvalue ( ) ) <EOL> return data <EOL> class ResponseHeaderDict ( HeaderDict ) : <EOL> def __init__ ( self , * args , ** kw ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> DeprecationWarning , <NUM_LIT:2> ) <EOL> HeaderDict . __init__ ( self , * args , ** kw ) <EOL> def _warn_deprecated ( new_func ) : <EOL> new_name = new_func . func_name <EOL> new_path = new_func . func_globals [ '<STR_LIT>' ] + '<STR_LIT:.>' + new_name <EOL> def replacement ( * args , ** kw ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> % ( new_name , new_path ) , <EOL> DeprecationWarning , <NUM_LIT:2> ) <EOL> return new_func ( * args , ** kw ) <EOL> try : <EOL> replacement . func_name = new_func . func_name <EOL> except : <EOL> pass <EOL> return replacement <EOL> for _name in __all__ : <EOL> _func = globals ( ) [ _name ] <EOL> if ( hasattr ( _func , '<STR_LIT>' ) <EOL> and _func . func_globals [ '<STR_LIT>' ] != __name__ ) : <EOL> globals ( ) [ _name ] = _warn_deprecated ( _func ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import doctest <EOL> doctest . testmod ( ) </s>
