1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""
There are attempts to create a new version of the WSGI standard. These
classes try to adapt the current standard to something that eventually
works out to be the next version of WSGI. For more information see
U{http://www.wsgi.org/wsgi/WSGI_2.0}.
"""
__all__ = []
from wsgitools.filters import CloseableIterator, CloseableList
try:
next
except NameError:
def next(it):
return it.next()
__all__.append("WSGI2to1Adapter")
class WSGI2to1Adapter:
"""Adapts an application with an interface that might somewhen be known as
WSGI 2.0 to the WSGI 1.0 interface."""
def __init__(self, app):
"""app is an application with an interface that might somewhen be known
as WSGI 2.0."""
self.app = app
def __call__(self, environ, start_response):
"""WSGI 1.0 interface"""
assert isinstance(environ, dict)
status, headers, iterable = self.app(environ)
assert isinstance(status, str)
assert isinstance(headers, list)
assert hasattr(iterable, "__iter__")
start_response(status, headers)
return iterable
__all__.append("WSGI1to2Adapter")
class WSGI1to2Adapter:
"""Adapts a WSGI 1.0 application to something that might somewhen be the
WSGI 2.0 interface."""
def __init__(self, app):
"""@param app: is a WSGI 1.0 application"""
self.app = app
def __call__(self, environ):
"""some interface that might somewhen be known as WSGI 2.0"""
assert isinstance(environ, dict)
results = [None, None, []]
def start_response(status, headers, exc_info=None):
assert isinstance(status, str)
assert isinstance(headers, list)
results[0] = status
results[1] = headers
def write_callable(data):
results[2].append(data)
return write_callable
iterable = self.app(environ, start_response)
assert hasattr(iterable, "__iter__")
if not results[2]:
return results[0], results[1], iterable
if isinstance(iterable, list):
# retaining .close attribute this way
iterable[:0] = results[2]
return results[0], results[1], iterable
close_function = getattr(iterable, "close", None)
iterable = iter(iterable)
try:
first = next(iterable)
except StopIteration:
return (results[0], results[1],
CloseableList(close_function, results[2]))
results[2].append(first)
return (results[0], results[1],
CloseableIterator(close_function, results[2], iterable))
|
