/
opt
/
alt
/
python37
/
lib
/
python3.7
/
site-packages
/
paste
/
/opt/alt/python37/lib/python3.7/site-packages/paste
mkdir
upload
Name
Size
Mode
Actions
auth/
-
0755
rm
cowbell/
-
0755
rm
debug/
-
0755
rm
evalexception/
-
0755
rm
exceptions/
-
0755
rm
util/
-
0755
rm
__pycache__/
-
0755
rm
cascade.py
4508
0644
edit
dl
rm
cascade.pyc
4475
0644
edit
dl
rm
cascade.pyo
4475
0644
edit
dl
rm
cgiapp.py
9590
0644
edit
dl
rm
cgiapp.pyc
8554
0644
edit
dl
rm
cgiapp.pyo
8337
0644
edit
dl
rm
cgitb_catcher.py
3752
0644
edit
dl
rm
cgitb_catcher.pyc
3851
0644
edit
dl
rm
cgitb_catcher.pyo
3851
0644
edit
dl
rm
config.py
4312
0644
edit
dl
rm
config.pyc
5284
0644
edit
dl
rm
config.pyo
5284
0644
edit
dl
rm
errordocument.py
13777
0644
edit
dl
rm
errordocument.pyc
13080
0644
edit
dl
rm
errordocument.pyo
13080
0644
edit
dl
rm
fileapp.py
13608
0644
edit
dl
rm
fileapp.pyc
13352
0644
edit
dl
rm
fileapp.pyo
13218
0644
edit
dl
rm
fixture.py
57992
0644
edit
dl
rm
fixture.py.stdlib
58075
0644
edit
dl
rm
fixture.pyc
60726
0644
edit
dl
rm
fixture.pyo
58453
0644
edit
dl
rm
flup_session.py
3941
0644
edit
dl
rm
flup_session.pyc
3959
0644
edit
dl
rm
flup_session.pyo
3959
0644
edit
dl
rm
gzipper.py
3692
0644
edit
dl
rm
gzipper.pyc
4578
0644
edit
dl
rm
gzipper.pyo
4578
0644
edit
dl
rm
httpexceptions.py
24222
0644
edit
dl
rm
httpexceptions.pyc
28380
0644
edit
dl
rm
httpexceptions.pyo
27623
0644
edit
dl
rm
httpheaders.py
43090
0644
edit
dl
rm
httpheaders.pyc
45274
0644
edit
dl
rm
httpheaders.pyo
43974
0644
edit
dl
rm
httpserver.py
55662
0644
edit
dl
rm
httpserver.pyc
48661
0644
edit
dl
rm
httpserver.pyo
47999
0644
edit
dl
rm
lint.py
14999
0644
edit
dl
rm
lint.pyc
17889
0644
edit
dl
rm
lint.pyo
13772
0644
edit
dl
rm
modpython.py
7976
0644
edit
dl
rm
modpython.pyc
8953
0644
edit
dl
rm
modpython.pyo
8953
0644
edit
dl
rm
pony.py
2279
0644
edit
dl
rm
pony.pyc
2753
0644
edit
dl
rm
pony.pyo
2753
0644
edit
dl
rm
progress.py
8162
0644
edit
dl
rm
progress.pyc
9416
0644
edit
dl
rm
progress.pyo
9416
0644
edit
dl
rm
proxy.py
10181
0644
edit
dl
rm
proxy.pyc
8787
0644
edit
dl
rm
proxy.pyo
8787
0644
edit
dl
rm
recursive.py
14664
0644
edit
dl
rm
recursive.pyc
18141
0644
edit
dl
rm
recursive.pyo
17816
0644
edit
dl
rm
registry.py
22193
0644
edit
dl
rm
registry.pyc
22941
0644
edit
dl
rm
registry.pyo
22941
0644
edit
dl
rm
reloader.py
6010
0644
edit
dl
rm
reloader.pyc
6134
0644
edit
dl
rm
reloader.pyo
6134
0644
edit
dl
rm
request.py
14145
0644
edit
dl
rm
request.pyc
14065
0644
edit
dl
rm
request.pyo
13969
0644
edit
dl
rm
response.py
7659
0644
edit
dl
rm
response.pyc
9468
0644
edit
dl
rm
response.pyo
9384
0644
edit
dl
rm
session.py
11335
0644
edit
dl
rm
session.pyc
12053
0644
edit
dl
rm
session.pyo
12053
0644
edit
dl
rm
transaction.py
4362
0644
edit
dl
rm
transaction.pyc
5857
0644
edit
dl
rm
transaction.pyo
5857
0644
edit
dl
rm
translogger.py
4812
0644
edit
dl
rm
translogger.pyc
4374
0644
edit
dl
rm
translogger.pyo
4374
0644
edit
dl
rm
url.py
14624
0644
edit
dl
rm
url.pyc
18581
0644
edit
dl
rm
url.pyo
18581
0644
edit
dl
rm
urlmap.py
9032
0644
edit
dl
rm
urlmap.pyc
9773
0644
edit
dl
rm
urlmap.pyo
9625
0644
edit
dl
rm
urlparser.py
26424
0644
edit
dl
rm
urlparser.pyc
22670
0644
edit
dl
rm
urlparser.pyo
22494
0644
edit
dl
rm
wsgilib.py
20134
0644
edit
dl
rm
wsgilib.pyc
22236
0644
edit
dl
rm
wsgilib.pyo
22180
0644
edit
dl
rm
wsgiwrappers.py
22165
0644
edit
dl
rm
wsgiwrappers.pyc
24450
0644
edit
dl
rm
wsgiwrappers.pyo
24450
0644
edit
dl
rm
Edit:
/opt/alt/python37/lib/python3.7/site-packages/paste/wsgilib.py
(20134B)
# (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 paste.request import get_cookies, parse_querystring, parse_formvars from paste.request import construct_url, path_info_split, path_info_pop from paste.response import HeaderDict, has_header, header_value, remove_header from paste.response import error_body_response, error_response, error_response_app from traceback import print_exception import urllib from cStringIO import StringIO import sys from urlparse import urlsplit import warnings __all__ = ['add_close', 'add_start_close', 'capture_output', 'catch_errors', 'catch_errors_app', 'chained_app_iters', 'construct_url', 'dump_environ', 'encode_unicode_app_iter', 'error_body_response', 'error_response', 'get_cookies', 'has_header', 'header_value', 'interactive', 'intercept_output', 'path_info_pop', 'path_info_split', 'raw_interactive', 'send_file'] class add_close(object): """ 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 self._closed = False def __iter__(self): return self def next(self): return self.app_iter.next() def close(self): self._closed = True if hasattr(self.app_iterable, 'close'): self.app_iterable.close() self.close_func() def __del__(self): if not self._closed: # We can't raise an error or anything at this stage print >> sys.stderr, ( "Error: app_iter.close() was not called when finishing " "WSGI request. finalization function %s not called" % self.close_func) class add_start_close(object): """ An an iterable that iterates over app_iter, calls start_func before the first item is returned, then calls close_func at the end. """ def __init__(self, app_iterable, start_func, close_func=None): self.app_iterable = app_iterable self.app_iter = iter(app_iterable) self.first = True self.start_func = start_func self.close_func = close_func self._closed = False def __iter__(self): return self def next(self): if self.first: self.start_func() self.first = False return self.app_iter.next() def close(self): self._closed = True if hasattr(self.app_iterable, 'close'): self.app_iterable.close() if self.close_func is not None: self.close_func() def __del__(self): if not self._closed: # We can't raise an error or anything at this stage print >> sys.stderr, ( "Error: app_iter.close() was not called when finishing " "WSGI request. finalization function %s not called" % self.close_func) class chained_app_iters(object): """ Chains several app_iters together, also delegating .close() to each of them. """ def __init__(self, *chained): self.app_iters = chained self.chained = [iter(item) for item in chained] self._closed = False def __iter__(self): return self def next(self): if len(self.chained) == 1: return self.chained[0].next() else: try: return self.chained[0].next() except StopIteration: self.chained.pop(0) return self.next() def close(self): self._closed = True got_exc = None for app_iter in self.app_iters: try: if hasattr(app_iter, 'close'): app_iter.close() except: got_exc = sys.exc_info() if got_exc: raise got_exc[0], got_exc[1], got_exc[2] def __del__(self): if not self._closed: # We can't raise an error or anything at this stage print >> sys.stderr, ( "Error: app_iter.close() was not called when finishing " "WSGI request. finalization function %s not called" % self.close_func) class encode_unicode_app_iter(object): """ Encodes an app_iterable's unicode responses as strings """ def __init__(self, app_iterable, encoding=sys.getdefaultencoding(), errors='strict'): self.app_iterable = app_iterable self.app_iter = iter(app_iterable) self.encoding = encoding self.errors = errors def __iter__(self): return self def next(self): content = self.app_iter.next() if isinstance(content, unicode): content = content.encode(self.encoding, self.errors) return content def close(self): if hasattr(self.app_iterable, 'close'): self.app_iterable.close() 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. """ 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. """ 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 is not None: ok_callback() return app_iter else: return _wrap_app_iter_app( environ, start_response, app_iter, error_callback_app, ok_callback, catch=catch) class _wrap_app_iter_app(object): def __init__(self, environ, start_response, app_iterable, error_callback_app, ok_callback, catch=Exception): 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 self.catch = catch 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.catch: 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='', raise_on_wsgi_error=False, **environ): """ Runs the application in a fake environment. """ assert "path_info" not in environ, "argument list changed" if raise_on_wsgi_error: errors = ErrorRaiser() else: 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)) path_info = urllib.unquote(path_info) # urlsplit returns unicode so coerce it back to str path_info, query = str(path_info), str(query) 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 = [] 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] 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.append app_iter = application(basic_environ, start_response) try: try: for s in app_iter: if not isinstance(s, str): raise ValueError( "The app_iter response can only contain str (not " "unicode); got: %r" % s) headers_sent.append(True) if not headers_set: raise AssertionError("Content sent w/o headers!") output.append(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'], ''.join(output), errors.getvalue()) class ErrorRaiser(object): def flush(self): pass def write(self, value): if not value: return raise AssertionError( "No errors should be written (got: %r)" % value) def writelines(self, seq): raise AssertionError( "No errors should be written (got lines: %s)" % list(seq)) def getvalue(self): return '' 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', str(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) from paste 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: .. code-block:: python 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, conditional=None, start_response=None): """ 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: .. code-block:: python def dehtmlifying_middleware(application): def replacement_app(environ, start_response): status, headers, body = intercept_output( environ, application) start_response(status, headers) 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 A third optional argument ``conditional`` should be a function that takes ``conditional(status, headers)`` and returns False if the request should not be intercepted. In that case ``start_response`` will be called and ``(None, None, app_iter)`` will be returned. You must detect that in your code and return the app_iter, like: .. code-block:: python def dehtmlifying_middleware(application): def replacement_app(environ, start_response): status, headers, body = intercept_output( environ, application, lambda s, h: header_value(headers, 'content-type').startswith('text/html'), start_response) if status is None: return body start_response(status, headers) body = re.sub(r'<.*?>', '', body) return [body] return replacement_app """ if conditional is not None and start_response is None: raise TypeError( "If you provide conditional you must also provide " "start_response") data = [] output = StringIO() def replacement_start_response(status, headers, exc_info=None): if conditional is not None and not conditional(status, headers): data.append(None) return start_response(status, headers, exc_info) if data: data[:] = [] data.append(status) data.append(headers) return output.write app_iter = application(environ, replacement_start_response) if data[0] is None: return (None, None, app_iter) 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.HeaderDict", 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()
Save
cmd:
run