| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600 |
- # (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.
- """
- from __future__ import print_function
- # 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 six
- import sys
- from six.moves import cStringIO as StringIO
- from six.moves.urllib.parse import unquote, 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("Error: app_iter.close() was not called when finishing "
- "WSGI request. finalization function %s not called"
- % self.close_func, file=sys.stderr)
- 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 next(self.app_iter)
- __next__ = 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("Error: app_iter.close() was not called when finishing "
- "WSGI request. finalization function %s not called"
- % self.close_func, file=sys.stderr)
- 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:
- six.reraise(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("Error: app_iter.close() was not called when finishing "
- "WSGI request. finalization function %s not called"
- % self.close_func, file=sys.stderr)
- 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 = next(self.app_iter)
- if isinstance(content, six.text_type):
- content = content.encode(self.encoding, self.errors)
- return content
- __next__ = next
- 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 = six.BytesIO()
- 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': six.BytesIO(),
- 'wsgi.errors': errors,
- 'wsgi.multithread': False,
- 'wsgi.multiprocess': False,
- 'wsgi.run_once': False,
- }
- if path:
- (_, _, path_info, query, fragment) = urlsplit(str(path))
- path_info = 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, bytes):
- basic_environ['wsgi.input'] = six.BytesIO(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
- six.reraise(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, six.binary_type):
- raise ValueError(
- "The app_iter response can only contain bytes (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 as 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'], b''.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 = list(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)
- if six.PY3:
- output = output.encode('utf8')
- 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()
|